时间戳

首页 > 解决方案 > 我怎样才能从 kafka 主题接收数据到我的 Streaming Structured DataFrame？

json - 我怎样才能从 kafka 主题接收数据到我的 Streaming Structured DataFrame？

问题描述

我知道如何使用我的 Kafka 主题中的数据，但我无法在正确的列中获取正确的数据。

我以这种格式接收列值中的所有数据：

{"timestamp":"2021-11-09T11:03:48.955+01:00","time":"1","duration":"0","SourceComputer":"C1707","SourcePort":"N1","DestinationComputer":"C925","start/end":" "}

有些字段是空的“”，而其他字段里面有一些数据（例如：“C1707”）。我在想我可以使用这个功能：

DataFrame=DataFrame.withColumn(ColumnName[i],split(DataFrame["value"],',').getItem(i))

但我无法在确切的列中获得确切的数据。

+--------------------+--------------------+----+--------+--------------+----------+-------------------+---------+
|               value|           timestamp|time|duration|SourceComputer|SourcePort|DestinationComputer|start/end|
+--------------------+--------------------+----+--------+--------------+----------+-------------------+---------+
|{"timestamp":"202...|{"timestamp":"202...|null|    null|          null|      null|               null|     null|
+--------------------+--------------------+----+--------+--------------+----------+-------------------+---------+

知道如何以正确的方式接收我的数据吗？

谢谢 ！

标签： jsonapache-sparkpysparkapache-kafkaspark-structured-streaming

解决方案

使用from_json方法，您可以将 Spark DataFrame 列上的 JSON 字符串转换为结构类型。然后你可以将你的结构类型转换为所需的数据框

import org.apache.spark.sql.functions.from_json
val schema = new StructType()
    .add("col1", StringType, true)
    .add("col2", StringType, true)
    .add("col3", StringType, true)
val df4=df.withColumn("value",from_json(col("value"),schema))
val df5=df4.select(col("value.*"))

请参考这个链接，这里解释了一切。

推荐阅读