我读过 Spark Structured Streaming 不支持将 Kafka 消息读取为 JSON 的模式推断。有没有办法像 Spark Streaming 一样检索架构：
val dataFrame = spark.read.json(rdd.map(_.value()))
dataFrame.printschema
【问题讨论】：
标签： apache-spark apache-kafka schema spark-structured-streaming
这是一种可能的方法：
在开始流式传输之前，从 Kafka 获取一小批数据
从小批量推断架构
使用提取的架构开始流式传输数据。
下面的伪代码说明了这种方法。
第 1 步：
从 Kafka 中提取一小批（两条记录），
val smallBatch = spark.read.format("kafka")
.option("kafka.bootstrap.servers", "node:9092")
.option("subscribe", "topicName")
.option("startingOffsets", "earliest")
.option("endingOffsets", """{"topicName":{"0":2}}""")
.load()
.selectExpr("CAST(value AS STRING) as STRING").as[String].toDF()
第 2 步： 将小批量写入文件：
smallBatch.write.mode("overwrite").format("text").save("/batch")
此命令将小批量写入 hdfs 目录 /batch。它创建的文件的名称是 part-xyz*。因此，您首先需要使用 hadoop FileSystem 命令重命名文件（参见 org.apache.hadoop.fs._ 和 org.apache.hadoop.conf.Configuration，这是一个示例 https://stackoverflow.com/a/41990859），然后将文件读取为 json:
val smallBatchSchema = spark.read.json("/batch/batchName.txt").schema
这里，batchName.txt 是文件的新名称，smallBatchSchema 包含从小批量推断的架构。
最后，您可以按如下方式流式传输数据（第 3 步）：
val inputDf = spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", "node:9092")
.option("subscribe", "topicName")
.option("startingOffsets", "earliest")
.load()
val dataDf = inputDf.selectExpr("CAST(value AS STRING) as json")
.select( from_json($"json", schema=smallBatchSchema).as("data"))
.select("data.*")
希望这会有所帮助！
【讨论】：
可以使用这个结构：
myStream = spark.readStream.schema(spark.read.json("my_sample_json_file_as_schema.json").schema).json("my_json_file")..
这怎么可能？好吧，由于 spark.read.json("..").schema 准确地返回了一个想要的推断模式，你可以使用这个返回的模式作为 spark.readStream 的强制模式参数的参数
我所做的是指定一个单行示例 json 作为推断架构内容的输入，这样就不会不必要地占用内存。如果您的数据发生变化，只需更新您的 sample-json。
我花了一些时间才弄明白（手动构建 StructTypes 和 StructFields 很痛苦..），因此我会为所有的赞成票感到高兴:-)
【讨论】：
这是不可能的。 Spark Streaming 在开发中支持有限的模式推断，spark.sql.streaming.schemaInference 设置为true：
默认情况下，来自基于文件的源的结构化流式处理要求您指定架构，而不是依靠 Spark 自动推断它。此限制可确保将一致的模式用于流式查询，即使在失败的情况下也是如此。对于临时用例，您可以通过将 spark.sql.streaming.schemaInference 设置为 true 来重新启用架构推断。
但它不能用于从 Kafka 消息中提取 JSON，并且 DataFrameReader.json 不支持将 Datasets 流式传输作为参数。
您必须手动提供架构How to read records in JSON format from Kafka using Structured Streaming?
【讨论】：
可以将 JSON 转换为 DataFrame，而无需手动输入架构，如果这是您要问的。
最近我遇到了一种情况，我通过 Kafka 接收大量长的嵌套 JSON 数据包，手动输入架构既麻烦又容易出错。
使用少量数据样本和一些技巧，您可以向 Spark2+ 提供架构，如下所示：
val jsonstr = """ copy paste a representative sample of data here"""
val jsondf = spark.read.json(Seq(jsonstr).toDS) //jsondf.schema has the nested json structure we need
val event = spark.readStream.format..option...load() //configure your source
val eventWithSchema = event.select($"value" cast "string" as "json").select(from_json($"json", jsondf.schema) as "data").select("data.*")
现在您可以像使用 Direct Streaming 一样使用此 val 做任何您想做的事情。创建临时视图，运行 SQL 查询，等等。
【讨论】：
将Arnon's 解决方案带到下一步（因为它在 spark 的较新版本中已被弃用，并且需要迭代整个数据框以进行类型转换）
spark.read.json(df.as[String])
无论如何，就目前而言，它仍处于试验阶段。