spark：使用Encoder生成DataFrame

62 阅读 0 评论 41 点赞

我是靠谱客的博主瘦瘦电脑，最近开发中收集的这篇文章主要介绍spark：使用Encoder生成DataFrame，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

基于spark，使用scala语言：

object Test {
    def main(args: Array[String]): Unit = {
        import org.apache.spark.sql.SparkSession
        val spark = SparkSession.builder().appName("encoder_test").master("local").getOrCreate()
        val m1 = Map("a" -> 1)
        val m2 = Map("b" -> 10)
        val m3 = Map("c" -> 100)
        val seq1 = Seq(m1, m2, m3)
        val a1 = Array(1, 2, 3)
        val a2 = Array(3, 5)
        val a3 = Array(1, 2, 3)
        val seq2 = Seq(a1, a2, a3)
        val s1 = "abc"
        val s2 = "abc2"
        val s3 = "abc3"
        val seq3 = Seq(s1, s2, s3)
        import org.apache.spark.sql.Row
        val r1 = Row(1, "a")
        val r2 = Row(2, "b")
        val r3 = Row(3, "c")
        val seq4 = Seq(r1, r2, r3)
        val seqt = seq1.zip(seq2).zipWithIndex.map { case (v, idx) => (v._1, v._2, seq3(idx)) }
        val seq = seqt.zipWithIndex.map { case (v, idx) => (v._1, v._2, v._3, seq4(idx)) }
        import spark.implicits._
        import org.apache.spark.sql.Encoder
        import org.apache.spark.sql.Encoders
        import org.apache.spark.sql.catalyst.encoders.RowEncoder
        import org.apache.spark.sql.types._
        val encoder = Encoders.tuple(
            newMapEncoder[Map[String, Int]],
            newIntArrayEncoder,
            Encoders.STRING,
            RowEncoder(
                StructType(
                    Seq(
                        StructField("num", IntegerType),
                        StructField("str", StringType)))))
        val d = spark.createDataset(seq)(encoder)
        d.printSchema()
        d.show()
    }
}

输出：

root
 |-- _1: map (nullable = true)
 |    |-- key: string
 |    |-- value: integer (valueContainsNull = false)
 |-- _2: array (nullable = true)
 |    |-- element: integer (containsNull = false)
 |-- _3: string (nullable = true)
 |-- _4: struct (nullable = true)
 |    |-- num: integer (nullable = true)
 |    |-- str: string (nullable = true)

+----------+---------+----+------+
|        _1|       _2|  _3|    _4|
+----------+---------+----+------+
|  [a -> 1]|[1, 2, 3]| abc|[1, a]|
| [b -> 10]|   [3, 5]|abc2|[2, b]|
|[c -> 100]|[1, 2, 3]|abc3|[3, c]|
+----------+---------+----+------+

说明：
其中import spark.implicits._返回一个继承SQLImplicits的对象，SQLImplicits里面实现了常见的Encoder，如上面用到了

implicit def newMapEncoder[T <: Map[_, _]](implicit arg0: scala.reflect.api.JavaUniverse.TypeTag[T]): Encoder[T]
implicit def newIntArrayEncoder: Encoder[Array[Int]]

分别实现了Map和Array[Int]的编码器。