我想从Kafka主题中读取数据,并创建spark tempview以按列分组?
+----+--------------------+
| key| value|
+----+--------------------+
|null|{"e":"trade","E":...|
|null|{"e":"trade","E":...|
|null|{"e":"trade","E":...|
但是我不能从tempview聚合数据??值列数据存储为字符串???
Dataset<Row> data = spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092,localhost:9093")
.option("subscribe", "data2-topic")
.option("startingOffsets", "latest")
.option ("group.id", "test")
.option("enable.auto.commit", "true")
.option("auto.commit.interval.ms", "1000")
.load();
data.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)");
data.createOrReplaceTempView("Tempdata");
data.show();
Dataset<Row> df2=spark.sql("SELECT e FROM Tempdata group by e");
df2.show();
1条答案
按热度按时间tyg4sfes1#
值列数据存储为字符串???
对。。因为你
CAST(value as STRING)
你会想用from_json
函数,该函数将行加载到您可以在其中搜索的适当Dataframe中。有关一些示例,请参见databrick关于kafka上结构化流媒体的博客
如果主要目标只是对一些字段进行分组,那么ksql可能是另一种选择。