Spark Streaming RDD转JSON（规避废弃API）：新方案无法解析JSON求建议

我来帮你梳理下问题所在和对应的解决办法：

首先，你之前用的旧方法（直接用read.json(rdd)）之所以能打印JSON但isStream=false，是因为这个API是把RDD当成静态的批处理数据来解析，本质上不是流式处理，而且在Spark 2.2.0后确实被标记为废弃，官方推荐用结构化流（Structured Streaming）的方式来处理流式JSON解析。

至于新方法无法解析JSON，大概率是这几个环节出了问题，你可以逐一排查：

1. 确保字节转字符串的编码正确 #

Kinesis返回的是Array[Byte]类型的消息，转字符串时一定要指定正确的编码（绝大多数场景是UTF-8），如果编码不对，JSON会变成乱码，自然无法解析。

比如把原来的：

val rowRDD = rdd.map(jstr => new String(jstr))

改成：

val rowRDD = rdd.map(bytes => new String(bytes, "UTF-8"))

2. 用结构化流原生对接Kinesis（推荐方案） #

Spark 2.x之后，结构化流是处理流式数据的标准方式，天生支持流式DataFrame（isStream=true），而且API更稳定。你可以直接用结构化流读取Kinesis数据，然后解析JSON，示例代码如下：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

// 初始化SparkSession
val spark = SparkSession.builder()
  .appName("KinesisJsonParser")
  .getOrCreate()
import spark.implicits._

// 定义完整的JSON Schema（把你之前没写完的字段补充完整）
val jsonSchema = StructType(Seq(
  StructField("clientTime", StringType, nullable = true),
  StructField("clientIp", StringType, nullable = true),
  // 这里添加其他需要解析的字段，比如userId、eventType等
))

// 读取Kinesis流式数据
val kinesisStreamDF = spark.readStream
  .format("kinesis")
  .option("streamName", "你的Kinesis流名称")
  .option("endpointUrl", "https://kinesis.你的区域.amazonaws.com")
  .option("initialPosition", "latest") // 或者"trim_horizon"读取最早数据
  .option("awsAccessKeyId", "你的AWS Access Key")
  .option("awsSecretKey", "你的AWS Secret Key")
  .load()

// 解析JSON：先把字节转成字符串，再用from_json解析成结构化数据
val parsedStreamDF = kinesisStreamDF
  .selectExpr("CAST(data AS STRING) AS jsonStr")
  .select(from_json(col("jsonStr"), jsonSchema).alias("eventData"))
  .select("eventData.*") // 展开所有字段

// 流式输出到控制台验证结果
val query = parsedStreamDF.writeStream
  .outputMode("append")
  .format("console")
  .option("truncate", false) // 完整显示内容
  .start()

query.awaitTermination()

3. 如果必须基于旧DStream转换（兼容方案） #

如果你暂时不想替换整个DStream架构，也可以把DStream转换成结构化的Streaming DataFrame，再用from_json解析，示例代码：

import org.apache.spark.streaming.StreamingContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

val spark = SparkSession.builder().getOrCreate()
import spark.implicits._
val ssc = new StreamingContext(spark.sparkContext, Seconds(5))

// 你的Kinesis DStream初始化逻辑
val kinesisStreams = // ... 这里是你原来创建Kinesis DStream的代码
val unionStreams = ssc.union(kinesisStreams)

// 定义完整的JSON Schema
val jsonSchema = StructType(Seq(
  StructField("clientTime", StringType, nullable = true),
  StructField("clientIp", StringType, nullable = true),
  // 补充其他字段
))

// 处理每个RDD，解析JSON
unionStreams.foreachRDD { (rdd, time) =>
  if (!rdd.isEmpty()) {
    // 字节转字符串（指定UTF-8编码）
    val jsonStrRDD = rdd.map(bytes => new String(bytes, "UTF-8"))
    // 转成DataFrame
    val jsonDF = jsonStrRDD.toDF("jsonStr")
    
    // 用from_json解析JSON
    val parsedDF = jsonDF
      .select(from_json(col("jsonStr"), jsonSchema).alias("data"))
      .select("data.*")
    
    // 打印验证
    println(s"===== 批次时间: ${time} =====")
    parsedDF.show(false)
  }
}

ssc.start()
ssc.awaitTermination()

4. 排查Schema匹配问题 #

这是最容易踩的坑：你定义的Schema必须和JSON消息的字段完全匹配，包括字段名大小写、数据类型。比如JSON里的字段是client_ip，但你Schema里写的是clientIp；或者JSON里clientTime是时间戳数字，但你Schema定义成StringType，都会导致解析失败，返回null或者空值。

你可以先把原始JSON字符串打印出来，对照着调整Schema的字段和类型。

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