结构化 Spark Streaming 多次写入

Question

我正在使用要写入 kafka 主题和 hbase 的数据流。 对于 Kafka，我使用如下格式：

dataset.selectExpr("id as key", "to_json(struct(*)) as value")
        .writeStream.format("kafka")
        .option("kafka.bootstrap.servers", Settings.KAFKA_URL)
        .option("topic", Settings.KAFKA_TOPIC2)
        .option("checkpointLocation", "/usr/local/Cellar/zookeepertmp")
        .outputMode(OutputMode.Complete())
        .start()

然后对于 Hbase，我会这样做：

  dataset.writeStream.outputMode(OutputMode.Complete())
    .foreach(new ForeachWriter[Row] {
      override def process(r: Row): Unit = {
        //my logic
      }

      override def close(errorOrNull: Throwable): Unit = {}

      override def open(partitionId: Long, version: Long): Boolean = {
        true
      }
    }).start().awaitTermination()

这会按预期写入 Hbase，但并不总是写入 kafka 主题。我不确定为什么会这样。

Answer 1

在火花中使用foreachBatch：

如果您想将流式查询的输出写入多个位置，那么您可以简单地多次写入输出 DataFrame/Dataset。但是，每次写入尝试都可能导致重新计算输出数据（包括可能重新读取输入数据）。为避免重新计算，您应该缓存输出 DataFrame/Dataset，将其写入多个位置，然后取消缓存。这是一个大纲。

streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    batchDF.persist()
    batchDF.write.format(…).save(…) // location 1
    batchDF.write.format(…).save(…) // location 2
    batchDF.unpersist()
}