使用 scala spark 应用自定义模式以从 rest api 发布响应 JSON答案

【问题标题】：Apply custom schema to post response JSON from rest api using scala spark使用 scala spark 应用自定义模式以从 rest api 发布响应 JSON
【发布时间】：2019-11-23 01:57:37
【问题描述】：

我已尝试发送 json 请求并接收 json 响应，并且我正在将 json 内容转换为 Dataframe，但架构是所有属性的字符串。无论如何，我的自定义架构是否应用于存储在中的 json 响应使用 scala 和 spark 的数据集。

val inputStream = entity.getContent()
content = scala.io.Source.fromInputStream(inputStream).getLines.mkString
inputStream.close
}
httpClient.getConnectionManager().shutdown()
println(content)

val rootelem = "data"
var JsonDF : Dataframe = null;
if (rootelem.equalsIgnoreCase("NULL"))
{
jsonDF = sqlContext.read.json(Seq(content).toDS)

}

else{jsonDF=sqlContext.read.json(Seq(content).toDS).select(explode(col(rootelem))。 as("child").select(col("child.*"))

jsonDF.show()
}}

【问题讨论】：

这可能会有所帮助stackoverflow.com/questions/47212966/…
@Prateek Prateek 这是针对 JSON 文件的，但在我的情况下，我直接从 rest post API 获得响应，我将其动态保存到数据框，但数据类型不匹配。对于文件级自定义架构适用，但获取响应并将其动态存储到数据框中与 JSON 文件的情况不匹配。

标签： scala apache-spark apache-spark-sql bigdata spark-structured-streaming

【解决方案1】：

创建您自己的 StructType 或必须动态创建 Struct 类型，然后使用您的 Json 和 structtype 创建数据框。示例：

 val schema = StructType(
  列表（
    StructField("num", IntegerType, true),
    StructField("字母", StringType, true)
  )
)

val df = spark.createDataFrame(
  spark.sparkContext.parallelize(数据),
  图式
)

请勾选此项以动态创建结构类型Dynamically build case class or schema

【讨论】：

这仅适用于静态文件\数据。但是对于动态收集 json 响应和应用架构不起作用请查看上面的代码。