在 Spark 结构化流中应用消息级别而不是数据帧级别的模式答案

【问题标题】：Applying schema at message level instead of dataframe level in Spark Structured Streaming在 Spark 结构化流中应用消息级别而不是数据帧级别的模式
【发布时间】：2021-06-25 16:48:59
【问题描述】：

所以，问题是因为我的架构可能依赖于 kafka 标头/密钥，我想在消息级别而不是数据帧级别应用架构。如何做到这一点？谢谢

为数据帧级别应用模式的代码 sn-p 是：

val ParsedDataFrame = kafkaStreamData.selectExpr("CAST(value AS STRING)", "CAST(key AS STRING)")
  .select(from_json(col("value"), Schema), col("key"))
  .select("value.*","key")

我想要类似的东西，

if(key == 'a'){
   use Schema1
}
else{
   use Schema2
}

P.S：我尝试使用 foreach 和 map 函数，但都不起作用，可能没有正确使用它们

【问题讨论】：

您必须首先使用过滤器将消息分离到单独的数据帧中。没有每行架构的概念

标签： scala apache-spark apache-kafka spark-structured-streaming

【解决方案1】：

由于数据类型不匹配，无法在同一行中应用不同的架构，因为您最终会得到 AnalysisException。

要对此进行测试，您可以进行以下实验。

在key:::value格式的Kafka主题中有以下数据：

a:::{"a":"foo","b":"bar"}
b:::{"a":"foo","b":"bar"}

在您的流式查询中定义：

val schemaA = new StructType().add("a", StringType)
val schemaB = new StructType().add("b", StringType)

val df = spark.readStream
  .format("kafka")
  .[...]
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .withColumn("parsedJson", 
    when(col("key") === "a", from_json(col("value"), schemaA))
    .otherwise(from_json(col("value"), schemaB)))

这将导致AnalysisException：

org.apache.spark.sql.AnalysisException: cannot resolve 'CASE WHEN (`key` = 'a') THEN jsontostructs(`value`) ELSE jsontostructs(`value`) END' due to data type mismatch: THEN and ELSE expressions should all be same type or coercible to a common type;;
'Project [key#21, value#22, CASE WHEN (key#21 = a) THEN jsontostructs(StructField(a,StringType,true), value#22) ELSE jsontostructs(StructField(b,StringType,true), value#22) END AS parsedJson#27]

正如@OneCricketeer 在 cmets 中提到的那样，您需要首先基于过滤器将 kafka 输入流分成几个 Dataframe，然后应用不同的模式来解析带有 json 字符串的列。

【讨论】：