如何在 foreachPartition 中将 Rows 转换为 JSON？

【问题标题】：How to convert Rows to JSON inside foreachPartition?如何在 foreachPartition 中将 Rows 转换为 JSON？
【发布时间】：2017-07-18 22:03:07
【问题描述】：

有没有办法在 foreachPartition 中将 Row 转换为 JSON？我看过 How to convert Row to json in Spark 2 Scala 。但是这种方法不起作用，因为我无法从 foreachPartition 中访问 sqlContext，而且我的数据包含嵌套类型。

 dataframe.foreachPartition { partitionOfRecords =>

    ..
    val connectionString: ConnectionStringBuilder = new ConnectionStringBuilder(
                eventHubsNamespace,
                eventHubName,
                policyName,
                policyKey)

    val eventHubsClient: EventHubClient = EventHubClient.createFromConnectionString(connectionString.toString()).get()

    val json = /* CONVERT partitionOfRecords to JSON */

    val bytes = json.getBytes()
    val eventData = new EventData(bytes)
    eventHubsClient.send(eventData)
  }

【问题讨论】：

标签： scala apache-spark dataframe apache-spark-sql

【解决方案1】：

我强烈建议在之前 foreachPartition 转换为 JSON。

原因是functions 对象中内置了对 JSON 的支持，您可以使用 to_json 函数构建“字符串化”JSON（无需恢复到相当复杂的编码）。

to_json(e: Column): Column 将包含StructType 或ArrayType 的StructTypes 的列转换为具有指定架构的JSON 字符串。

我建议执行以下操作：

dataframe.
  select(to_json($"your-struct-column-here")).
  as[String].
  foreachPartition { json: String => ... }

【讨论】：