如何用Spark Streaming处理Kafka Topic读取的DataFrame并解析内容

一、将Kafka Value转为可读字符串 #

Kafka存储的Twitter数据默认是字节数组格式，Spark消费后返回BinaryType类型，直接打印会显示不可识别内容，只需将其强制转换为字符串即可：

• Scala代码：

  // kafkaDF为消费Kafka得到的原始DataFrame
  val readableDF = kafkaDF.selectExpr("CAST(value AS STRING) AS tweet_content")
  

• Python代码：

  readableDF = kafkaDF.selectExpr("CAST(value AS STRING) AS tweet_content")
  

如果转字符串后仍乱码，检查生产者写入Kafka时是否使用UTF-8编码，确保生产消费端编码一致。

二、处理解析后的Twitter DataFrame #

Twitter数据通常为JSON格式，转成字符串后可解析为结构化数据进行后续处理：

1. 解析JSON结构 #

先定义推文的Schema（根据实际返回的Twitter字段调整），再用from_json函数解析：

• Scala示例：

  import org.apache.spark.sql.functions.from_json
  import org.apache.spark.sql.types.StructType
  
  // 自定义推文Schema
  val tweetSchema = new StructType()
    .add("text", "string")
    .add("created_at", "string")
    .add("user", new StructType().add("screen_name", "string"))
  
  val parsedDF = readableDF.select(from_json($"tweet_content", tweetSchema).alias("tweet"))
    .select("tweet.text", "tweet.created_at", "tweet.user.screen_name")
  

• Python示例：

  from pyspark.sql.functions import from_json
  from pyspark.sql.types import StructType, StructField, StringType
  
  tweet_schema = StructType([
      StructField("text", StringType()),
      StructField("created_at", StringType()),
      StructField("user", StructType([StructField("screen_name", StringType())]))
  ])
  
  parsed_df = readableDF.select(from_json("tweet_content", tweet_schema).alias("tweet"))
    .select("tweet.text", "tweet.created_at", "tweet.user.screen_name")
  

2. 流式处理操作 #

解析完成后可进行过滤、统计等业务逻辑，比如过滤含特定关键词的推文并输出到控制台：

• Scala代码：

  val filteredDF = parsedDF.filter($"text".contains("大数据"))
  filteredDF.writeStream
    .outputMode("append")
    .format("console")
    .start()
    .awaitTermination()
  

• Python代码：

  filtered_df = parsed_df.filter(parsed_df["text"].contains("大数据"))
  filtered_df.writeStream \
    .outputMode("append") \
    .format("console") \
    .start() \
    .awaitTermination()
  

3. 版本兼容注意 #

确保Spark与Kafka版本匹配（如Spark 3.x对应Kafka 2.0及以上版本），避免因版本不兼容导致的解析异常。

内容的提问来源于stack exchange，提问作者Kulasangar