时间戳

首页 > 解决方案 > 带有 Json 输出的 Spark 数据集以 Kryo 编码形式出现

apache-spark - 带有 Json 输出的 Spark 数据集以 Kryo 编码形式出现

问题描述

我想将 CSV 数据转换为特定的 JSON 格式，而不是通常的平面格式。所以我尝试了以下方法：

case class CityInfo(lat : Double, long : Double)
  case class City(name: String, country: String, info : CityInfo)


    implicit def kryoEncoder[A](implicit ct: ClassTag[A]) =
      org.apache.spark.sql.Encoders.[A](ct)

    val cities = spark.read.option("header", true).csv("src/main/resources/worldcities.csv").cache()

    cities.map {
      case Row(city: String, _: Any, latStr : String, long : String, country: String, _*) =>
        City(city, country, CityInfo(latStr.toDouble, long.toDouble))
    }.write.json("data/testfile.json")

由此，我希望 json 行带有嵌套对象（如 City 和 CityInfo）。但我得到如下行：

{“值”：“AQBjb20uaGIuZXhlcmNpc2UuU3AxJENpdPkBAUtvc2927wEBY29tLmhiLmV4ZXJjaXNlLlNwMSRDaXR5SW5m7wFARXULDye7MEA0sJAt4A0bAVp1YmluIFBvdG/r”}

我使用了 Kryo 编码器，因为它无法编译，所以我找到了关于它的堆栈溢出答案……不确定它是否正确。

标签： apache-sparkapache-spark-datasetkryo

解决方案

我自己也很快找到了解决方案。因此，如果您使用的是 spark-shell 或类似的，您将拥有所有的隐含功能，并且不需要任何特别的东西。但是，如果像我一样，您正在从头开始编写 scala 应用程序，则需要添加：

import spark.implicits._

这将添加隐式编码器。还要删除 Kryo 的东西。注意：只要（在 CityInfo 中）由于是保留字而产生错误，上述示例就不会起作用。

推荐阅读