【问题标题】:How to create a Dataset of Maps?如何创建地图数据集?
【发布时间】:2018-03-28 11:12:49
【问题描述】:

我正在使用 Spark 2.2,但在尝试通过 SeqMap 调用 spark.createDataset 时遇到了麻烦。

我的 Spark Shell 会话的代码和输出如下:

// createDataSet on Seq[T] where T = Int works
scala> spark.createDataset(Seq(1, 2, 3)).collect
res0: Array[Int] = Array(1, 2, 3)

scala> spark.createDataset(Seq(Map(1 -> 2))).collect
<console>:24: error: Unable to find encoder for type stored in a Dataset.  
Primitive types (Int, String, etc) and Product types (case classes) are 
supported by importing spark.implicits._
Support for serializing other types will be added in future releases.
       spark.createDataset(Seq(Map(1 -> 2))).collect
                          ^

// createDataSet on a custom case class containing Map works
scala> case class MapHolder(m: Map[Int, Int])
defined class MapHolder

scala> spark.createDataset(Seq(MapHolder(Map(1 -> 2)))).collect
res2: Array[MapHolder] = Array(MapHolder(Map(1 -> 2)))

我已经尝试过import spark.implicits._,但我相当确定这是由 Spark shell 会话隐式导入的。

这是当前编码器未涵盖的情况吗?

【问题讨论】:

    标签: scala apache-spark apache-spark-sql apache-spark-dataset apache-spark-encoders


    【解决方案1】:

    2.2 中没有涉及,但可以轻松解决。您可以使用ExpressionEncoder 明确添加所需的Encoder

    import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder  
    import org.apache.spark.sql.Encoder
    
    spark
      .createDataset(Seq(Map(1 -> 2)))(ExpressionEncoder(): Encoder[Map[Int, Int]])
    

    implicitly:

    implicit def mapIntIntEncoder: Encoder[Map[Int, Int]] = ExpressionEncoder()
    spark.createDataset(Seq(Map(1 -> 2)))
    

    【讨论】:

      【解决方案2】:

      仅供参考,上面的表达式只适用于 Spark 2.3(如果我没记错的话,截至 this commit)。

      scala> spark.version
      res0: String = 2.3.0
      
      scala> spark.createDataset(Seq(Map(1 -> 2))).collect
      res1: Array[scala.collection.immutable.Map[Int,Int]] = Array(Map(1 -> 2))
      

      我认为这是因为 newMapEncoder 现在是 spark.implicits 的一部分。

      scala> :implicits
      ...
        implicit def newMapEncoder[T <: scala.collection.Map[_, _]](implicit evidence$3: reflect.runtime.universe.TypeTag[T]): org.apache.spark.sql.Encoder[T]
      

      您可以使用以下技巧“禁用”隐式并尝试上述表达式(这将导致错误)。

      trait ThatWasABadIdea
      implicit def newMapEncoder(ack: ThatWasABadIdea) = ack
      
      scala> spark.createDataset(Seq(Map(1 -> 2))).collect
      <console>:26: error: Unable to find encoder for type stored in a Dataset.  Primitive types (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._  Support for serializing other types will be added in future releases.
             spark.createDataset(Seq(Map(1 -> 2))).collect
                                ^
      

      【讨论】:

        猜你喜欢
        • 1970-01-01
        • 2017-01-10
        • 2018-12-03
        • 2019-07-02
        • 2018-01-05
        • 2018-10-19
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        相关资源
        最近更新 更多