结构化 Spark Streaming：writeStream 显示空数据帧答案

【问题标题】：structred Spark Streaming : writeStream display null dataframe结构化 Spark Streaming：writeStream 显示空数据帧
【发布时间】：2021-11-18 13:46:51
【问题描述】：

我的任务是使用 structred spark streaming python 进行实时处理所以第一步是将 csv 文件摄取到 kafka 主题中：完成

第二步是从主题kafka中读取流

    df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
  .option("subscribe", kafka_topic_name) \
  .option("startingOffsets", "latest")\
  .load()

然后我使用模式来转换我的数据框的列

from pyspark.sql.types import *

schema = StructType() \
      .add("DriverId",IntegerType(),True) \
      .add("time",TimestampType(),True) \
      .add("Longitude",DoubleType(),True) \
      .add("Latitude",DoubleType(),True) \
      .add("SPEED",DoubleType(),True) \
      .add("EngineSpeed",IntegerType(),True) \
      .add("MAF",IntegerType(),True) \
      .add("FuelType",IntegerType(),True) \

第二步是在控制台上显示我正在播放的内容，看看我是否走对了：

query1 = df\
    .writeStream\
    .format("console")\
    .outputMode("append")\
    .option("truncate", False)\
    .start()\
    .awaitTermination()

但结果是一个带有空值的数据框

所以，我已返回架构并将所有列转换为 stringType 结果是：

类似 json 的东西！

**我的问题是**如何正确转换我的数据框以及如何在每列下显示值和注释，如格式 json

【问题讨论】：

标签： csv apache-spark pyspark apache-kafka

【解决方案1】：

当 spark 从 kafka 读取数据时，它会创建一个包含 2 列的数据帧 - 键和值（这些对应于您发送给 kafka 的键和值
这2列的初始数据类型是ByteType

您必须将它们转换为 StringType

df = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

然后，假设你的 value 列是一个 json 字符串，你可以将它转换为你指定的 schema 的 struct 类型
```
df = df.select(F.from_json(df.value,schema=my_schema).alias("value"))
df = df.selectExpr("value.*")
```

更多信息可here

【讨论】：

就在我从 kafka 读取流时，整个正确的代码是 df = spark \ .readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \ .option ("subscribe", kafka_topic_name) \ .option("startingOffsets", "latest")\ .load() \ .selectExpr("CAST(value AS STRING)") \ .select(from_json(col("value"), schema).alias("data")) \ .select("data.*")
@girl_of_data 好东西。如果它对您有用，请用绿色勾号接受我的回答。如果你这样做，我们都会赢得声誉:)