如何处理 JSON 文档（来自 MongoDB）并在结构化流中写入 HBase？

Question

我正在获取 mongoDB 文档，然后在处理后我想使用 Bson.Document 库将其存储到 Hbase 中

将流式传输方法从 spark kafkastreaming 更改为结构化流式传输 所以早期使用 kafkaUtils 的方法是产生 Dstream[Document]

在结构化流中我得到数据集[文档]

scala> val stream = spark.readStream.format("kafka").option("kafka.bootstrap.servers","brokerList").option("subscribe", s"topic_name").load().selectExpr("CAST(value AS STRING)")
stream: org.apache.spark.sql.DataFrame = [value: string]

scala> val strming_doc = stream.map(record => record.getAs[String]("value")
scala> org.apache.spark.sql.Dataset[String] = [value: string]

对于进一步的过程，我需要从数据集中获取文档

scala> val stream = spark.readStream.format("kafka").option("kafka.bootstrap.servers","brokerList").option("subscribe", s"topic_name").load().selectExpr("CAST(value AS STRING)")
stream: org.apache.spark.sql.DataFrame = [value: string]

scala> val strming_doc = stream.map(record => record.getAs[String]("value")
scala> org.apache.spark.sql.Dataset[String] = [value: string]

我需要从数据集中获取文档，基本上是从 mongoDB 获取数据

Answer 1

看来您需要 foreach 或 foreachBatch 运算符将流式查询的结果写入 HBase。请参考官方文档中的Using Foreach and ForeachBatch。