apache-spark - 如何读取 Spark 中的嵌套集合

Question

我有一个 Parquet 表，其中一列是

, array<struct<col1,col2,..colN>>

可以使用 LATERAL VIEW 语法在 Hive 中针对此表运行查询。

如何将此表读入 RDD，更重要的是如何在 Spark 中过滤、映射此嵌套集合？

在 Spark 文档中找不到对此的任何引用。提前感谢您提供任何信息!

附言。我觉得在桌面上提供一些统计数据可能会有所帮助。 主表中的列数 ~600。行数 ~200m。 嵌套集合中的“列”数 ~10。嵌套集合中的平均记录数约为 35。

Answer 1

嵌套集合没有什么神奇之处。 Spark 将以相同的方式处理 RDD[(String, String)] 和 RDD[(String, Seq[String])]。

不过，从 Parquet 文件中读取此类嵌套集合可能很棘手。

让我们以 spark-shell (1.3.1) 为例:

scala> import sqlContext.implicits._
import sqlContext.implicits._

scala> case class Inner(a: String, b: String)
defined class Inner

scala> case class Outer(key: String, inners: Seq[Inner])
defined class Outer

编写parquet文件:

scala> val outers = sc.parallelize(List(Outer("k1", List(Inner("a", "b")))))
outers: org.apache.spark.rdd.RDD[Outer] = ParallelCollectionRDD[0] at parallelize at <console>:25

scala> outers.toDF.saveAsParquetFile("outers.parquet")

读取拼花文件:

scala> import org.apache.spark.sql.catalyst.expressions.Row
import org.apache.spark.sql.catalyst.expressions.Row

scala> val dataFrame = sqlContext.parquetFile("outers.parquet")
dataFrame: org.apache.spark.sql.DataFrame = [key: string, inners: array<struct<a:string,b:string>>]   

scala> val outers = dataFrame.map { row =>
     |   val key = row.getString(0)
     |   val inners = row.getAs[Seq[Row]](1).map(r => Inner(r.getString(0), r.getString(1)))
     |   Outer(key, inners)
     | }
outers: org.apache.spark.rdd.RDD[Outer] = MapPartitionsRDD[8] at map at DataFrame.scala:848

重要的部分是row.getAs[Seq[Row]](1)。 struct 的嵌套序列的内部表示是 ArrayBuffer[Row]，您可以使用它的任何父类(super class)型代替 Seq[Row]。 1 是外行中的列索引。我在这里使用了 getAs 方法，但在最新版本的 Spark 中有替代方法。查看 Row trait 的源代码.

现在您有了一个 RDD[Outer]，您可以应用任何想要的转换或操作。

// Filter the outers
outers.filter(_.inners.nonEmpty)

// Filter the inners
outers.map(outer => outer.copy(inners = outer.inners.filter(_.a == "a")))

请注意，我们仅使用 spark-SQL 库来读取 parquet 文件。例如，您可以直接在 DataFrame 上只选择需要的列，然后再将其映射到 RDD。

dataFrame.select('col1, 'col2).map { row => ... }