我们正在对从MySQL收集的kafka数据进行流处理。现在,一旦完成所有分析工作,我希望将数据直接保存到Hbase。我已经阅读了Spark结构化流文档,但没有找到任何与Hbase配合使用的sink。以下是我用于从Kafka读取数据的代码。
现在最终,我想要将DF_Hbase数据框保存在hbase中。
val records = spark.readStream.format("kafka").option("subscribe", "kaapociot").option("kafka.bootstrap.servers", "XX.XX.XX.XX:6667").option("startingOffsets", "earliest").load
val jsonschema = StructType(Seq(StructField("header", StringType, true),StructField("event", StringType, true)))
val uschema = StructType(Seq(
StructField("MeterNumber", StringType, true),
StructField("Utility", StringType, true),
StructField("VendorServiceNumber", StringType, true),
StructField("VendorName", StringType, true),
StructField("SiteNumber", StringType, true),
StructField("SiteName", StringType, true),
StructField("Location", StringType, true),
StructField("timestamp", LongType, true),
StructField("power", DoubleType, true)
))
val DF_Hbase = records.selectExpr("cast (value as string) as Json").select(from_json($"json",schema=jsonschema).as("data")).select("data.event").select(from_json($"event", uschema).as("mykafkadata")).select("mykafkadata.*")
现在最终,我想要将DF_Hbase数据框保存在hbase中。