Spark SQL AnalysisException：无法解析product列问题排查

从你贴出的日志和错误信息来看，问题很明确：你的代码试图访问product列，但Spark当前能识别的输入列只有jsontostructs(message)——这说明你已经把Kafka的message解析成了JSON结构体，但没有正确处理这个结构体，导致无法直接引用内部的product字段。

我帮你梳理下具体的解决思路和代码示例：

问题根源 #

你大概率是用jsontostructs（或者from_json）函数解析了Kafka的message，但没有给解析后的结构体列起别名，也没有用正确的方式访问结构体内部的字段。比如如果你的代码是这样的：

Dataset<Row> df = kafkaDF.select(functions.jsontostructs(functions.col("message")));
df.select("product").show();

必然会触发这个错误，因为此时DataFrame的列是jsontostructs(message)，而不是展开的product字段。

解决方案1：给结构体列起别名，用点符号访问字段 #

先给解析后的结构体列一个明确的别名（比如data），然后通过别名.字段名的方式访问product：

// 解析JSON并给结构体列起别名
Dataset<Row> parsedDF = kafkaDF.select(functions.jsontostructs(functions.col("message")).alias("data"));
// 访问结构体内部的product字段
Dataset<Row> resultDF = parsedDF.select("data.product");

解决方案2：直接展开结构体所有字段 #

如果你的JSON里有多个字段，想一次性展开所有字段，可以用select("别名.*")，之后就能直接引用product了：

Dataset<Row> parsedDF = kafkaDF.select(functions.jsontostructs(functions.col("message")).alias("data"));
// 展开结构体的所有字段
Dataset<Row> flattenedDF = parsedDF.select("data.*");
// 此时可以直接访问product列
flattenedDF.select("product").show();

额外排查要点 #

1. 确认JSON里确实有product字段：可以先打印解析后的结构体内容，验证字段是否存在：

   parsedDF.show(false);
   

2. 显式指定JSON Schema（更可靠）：Spark 2.2+推荐用from_json函数并显式定义Schema，避免自动解析出错：

   // 根据你的实际JSON结构定义Schema
   StructType jsonSchema = new StructType()
       .add("product", StringType)
       .add("price", DoubleType)
       .add("quantity", IntegerType);
   
   Dataset<Row> parsedDF = kafkaDF.select(functions.from_json(functions.col("message"), jsonSchema).alias("data"));
   

按照上面的方法调整代码后，应该就能解决这个解析错误了。

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