blocks|key|135387|text|在嵌套集合的情况下没有魔术。Spark将以与RDD[(String,+String)]和RDD[(String,+Seq[String])]相同的方式处理。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|135388|不过，从Parquet文件中读取此类嵌套集合可能会比较棘手。|135389|让我们以spark-shell+(1.3.1)为例：|135390|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|code-block|syntax|javascript|135391|编写拼图文件：|135392|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")|135393|阅读拼图文件：|135394|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+=>
+++++%7C+++val+key+=+row.getString(0)
+++++%7C+++val+inners+=+row.getAs[Seq[Row]](1).map(r+=>+Inner(r.getString(0),+r.getString(1)))
+++++%7C+++Outer(key,+inners)
+++++%7C+}
outers:+org.apache.spark.rdd.RDD[Outer]+=+MapPartitionsRDD[8]+at+map+at+DataFrame.scala:848|135395|最重要的部分是row.getAs[Seq[Row]](1)。struct嵌套序列的内部表示是ArrayBuffer[Row]，您可以使用它的任何超类型来代替Seq[Row]。1是外部行中的列索引。我在这里使用了getAs方法，但在最新版本的Spark中也有替代方法。请参阅Row+trait的源代码。|135396|现在您有了一个RDD[Outer]，您可以应用任何所需的转换或操作。|135397|//+Filter+the+outers
outers.filter(_.inners.nonEmpty)

//+Filter+the+inners
outers.map(outer+=>+outer.copy(inners+=+outer.inners.filter(_.a+==+"a")))|135398|请注意，我们使用spark-SQL库仅用于读取parquet文件。例如，您可以直接在DataFrame上选择所需的列，然后再将其映射到RDD。|135399|dataFrame.select('col1,+'col2).map+{+row+=>+...+}|135400|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/Row.scala#L118^0|M|L|18|Q|0|0|4|B|0|0|0|0|0|0|7|M|U|6|1A|G|26|8|2F|1|2X|5|3S|9|0|0|7|A|0|0|0|0^^$0|@$1|2|3|4|5|6|7|1E|8|@$9|1F|A|1G|B|C]|$9|1H|A|1I|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|1J|8|@]|D|@]|E|$]]|$1|H|3|I|5|6|7|1K|8|@$9|1L|A|1M|B|C]]|D|@]|E|$]]|$1|J|3|K|5|L|7|1N|8|@]|D|@]|E|$M|N]]|$1|O|3|P|5|6|7|1O|8|@]|D|@]|E|$]]|$1|Q|3|R|5|L|7|1P|8|@]|D|@]|E|$M|N]]|$1|S|3|T|5|6|7|1Q|8|@]|D|@]|E|$]]|$1|U|3|V|5|L|7|1R|8|@]|D|@]|E|$M|N]]|$1|W|3|X|5|6|7|1S|8|@$9|1T|A|1U|B|C]|$9|1V|A|1W|B|C]|$9|1X|A|1Y|B|C]|$9|1Z|A|20|B|C]|$9|21|A|22|B|C]|$9|23|A|24|B|C]]|D|@$9|25|A|26|1|27]]|E|$]]|$1|Y|3|Z|5|6|7|28|8|@$9|29|A|2A|B|C]]|D|@]|E|$]]|$1|10|3|11|5|L|7|2B|8|@]|D|@]|E|$M|N]]|$1|12|3|13|5|6|7|2C|8|@]|D|@]|E|$]]|$1|14|3|15|5|L|7|2D|8|@]|D|@]|E|$M|N]]|$1|16|3|-4|5|6|7|2E|8|@]|D|@]|E|$]]]|17|$18|$5|19|1A|1B|E|$1C|1D]]]]

There is no magic in the case of nested collection. Spark will handle the same way a <code>RDD[(String, String)]</code> and a <code>RDD[(String, Seq[String])]</code>. 

Reading such nested collection from Parquet files can be tricky, though. 

Let's take an example from the <code>spark-shell</code> (1.3.1): 

<pre><code>scala&gt; import sqlContext.implicits._
import sqlContext.implicits._

scala&gt; case class Inner(a: String, b: String)
defined class Inner

scala&gt; case class Outer(key: String, inners: Seq[Inner])
defined class Outer
</code></pre>

Write the parquet file: 

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

scala&gt; outers.toDF.saveAsParquetFile("outers.parquet")
</code></pre>

Read the parquet file:

<pre><code>scala&gt; import org.apache.spark.sql.catalyst.expressions.Row
import org.apache.spark.sql.catalyst.expressions.Row

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

scala&gt; val outers = dataFrame.map { row =&gt;
 | val key = row.getString(0)
 | val inners = row.getAs[Seq[Row]](1).map(r =&gt; 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
</code></pre>

The important part is <code>row.getAs[Seq[Row]](1)</code>. The internal representation of a nested sequence of <code>struct</code> is <code>ArrayBuffer[Row]</code>, you could use any super-type of it instead of <code>Seq[Row]</code>. The <code>1</code> is the column index in the outer row. I used the method <code>getAs</code> here but there are alternatives in the latest versions of Spark. See the source code of the <a href="https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/Row.scala#L118" rel="noreferrer">Row trait</a>.

Now that you have a <code>RDD[Outer]</code>, you can apply any wanted transformation or action.

<pre><code>// Filter the outers
outers.filter(_.inners.nonEmpty)

// Filter the inners
outers.map(outer =&gt; outer.copy(inners = outer.inners.filter(_.a == "a")))
</code></pre>

Note that we used the spark-SQL library only to read the parquet file. You could for example select only the wanted columns directly on the DataFrame, before mapping it to a RDD.

<pre><code>dataFrame.select('col1, 'col2).map { row =&gt; ... }
</code></pre>

blocks|key|137306|text|我将给出一个基于Python的答案，因为这就是我正在使用的。我认为Scala也有类似的东西。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|137307|根据Python+API+docs的说法，在Spark+1.4.0中添加了explode函数，以处理DataFrames中的嵌套数组。|offset|length|style|CODE|137308|创建测试数据帧：|137309|from+pyspark.sql+import+Row

df+=+sqlContext.createDataFrame([Row(a=1,+intlist=[1,2,3]),+Row(a=2,+intlist=[4,5,6])])
df.show()

##+%2B-%2B--------------------%2B
##+%7Ca%7C+++++++++++++intlist%7C
##+%2B-%2B--------------------%2B
##+%7C1%7CArrayBuffer(1,+2,+3)%7C
##+%7C2%7CArrayBuffer(4,+5,+6)%7C
##+%2B-%2B--------------------%2B|code-block|syntax|javascript|137310|使用explode平整列表列：|137311|from+pyspark.sql.functions+import+explode

df.select(df.a,+explode(df.intlist)).show()

##+%2B-%2B---%2B
##+%7Ca%7C_c0%7C
##+%2B-%2B---%2B
##+%7C1%7C++1%7C
##+%7C1%7C++2%7C
##+%7C1%7C++3%7C
##+%7C2%7C++4%7C
##+%7C2%7C++5%7C
##+%7C2%7C++6%7C
##+%2B-%2B---%2B|137312|entityMap|0|LINK|mutability|MUTABLE|url|https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=explode#pyspark.sql.functions.explode^0|0|11|7|2|F|0|0|0|0|2|7|0|0^^$0|@$1|2|3|4|5|6|7|10|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|11|8|@$D|12|E|13|F|G]]|9|@$D|14|E|15|1|16]]|A|$]]|$1|H|3|I|5|6|7|17|8|@]|9|@]|A|$]]|$1|J|3|K|5|L|7|18|8|@]|9|@]|A|$M|N]]|$1|O|3|P|5|6|7|19|8|@$D|1A|E|1B|F|G]]|9|@]|A|$]]|$1|Q|3|R|5|L|7|1C|8|@]|9|@]|A|$M|N]]|$1|S|3|-4|5|6|7|1D|8|@]|9|@]|A|$]]]|T|$U|$5|V|W|X|A|$Y|Z]]]]

I'll give a Python-based answer since that's what I'm using. I think Scala has something similar.

The <code>explode</code> function was added in Spark 1.4.0 to handle nested arrays in DataFrames, according to the <a href="https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=explode#pyspark.sql.functions.explode">Python API docs</a>.

Create a test dataframe:

<pre><code>from pyspark.sql import Row

df = sqlContext.createDataFrame([Row(a=1, intlist=[1,2,3]), Row(a=2, intlist=[4,5,6])])
df.show()

## +-+--------------------+
## |a| intlist|
## +-+--------------------+
## |1|ArrayBuffer(1, 2, 3)|
## |2|ArrayBuffer(4, 5, 6)|
## +-+--------------------+
</code></pre>

Use <code>explode</code> to flatten the list column:

<pre><code>from pyspark.sql.functions import explode

df.select(df.a, explode(df.intlist)).show()

## +-+---+
## |a|_c0|
## +-+---+
## |1| 1|
## |1| 2|
## |1| 3|
## |2| 4|
## |2| 5|
## |2| 6|
## +-+---+
</code></pre>

blocks|key|2703267|text|另一种方法是像这样使用模式匹配：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|2703268|val+rdd:+RDD[(String,+List[(String,+String)]]+=+dataFrame.map(_.toSeq.toList+match+{+
++case+List(key:+String,+inners:+Seq[Row])+=>+key+->+inners.map(_.toSeq.toList+match+{
++++case+List(a:String,+b:+String)+=>+(a,+b)
++}).toList
})|code-block|syntax|javascript|2703269|您可以直接在Row上进行模式匹配，但由于几个原因，它可能会失败。|2703270|entityMap^0|0|0|0^^$0|@$1|2|3|4|5|6|7|K|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|L|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|M|8|@]|9|@]|A|$]]|$1|I|3|-4|5|6|7|N|8|@]|9|@]|A|$]]]|J|$]]

Another approach would be using pattern matching like this:

<pre><code>val rdd: RDD[(String, List[(String, String)]] = dataFrame.map(_.toSeq.toList match { 
 case List(key: String, inners: Seq[Row]) =&gt; key -&gt; inners.map(_.toSeq.toList match {
 case List(a:String, b: String) =&gt; (a, b)
 }).toList
})
</code></pre>

You can pattern match directly on Row but it is likely to fail for a few reasons.

blocks|key|137328|text|上面的答案都是很好的答案，从不同的角度解决了这个问题；Spark+SQL也是访问嵌套数据的一种非常有用的方式。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|137329|下面是如何在SQL中直接使用explode()来查询嵌套集合的示例。|137330|SELECT+hholdid,+tsp.person_seq_no+
FROM+(++SELECT+hholdid,+explode(tsp_ids)+as+tsp+
++++++++FROM+disc_mrt.unified_fact+uf
+++++)|code-block|syntax|javascript|137331|tsp_ids是一个嵌套的结构，它有许多属性，包括我在上面的外部查询中选择的person_seq_no。|137332|上面的测试是在Spark+2.0上进行的。我做了一个小测试，它在Spark+1.6中不起作用。这个问题是在Spark+2不在的时候提出的，所以这个答案很好地添加到了处理嵌套结构的可用选项列表中。|137333|看看下面的JIRAs+for+Hive兼容的使用LATERAL+VIEW+OUTER语法查询嵌套数据的方法，因为Spark2.2也支持OUTER爆炸(例如，当嵌套集合为空，但您仍然希望拥有来自父记录的属性时)：|offset|length|style|CODE|137334|137335|SPARK-13721:+Add+support+for+LATERAL+VIEW+OUTER+explode()|unordered-list-item|137336|137337|对于SQL访问，explode()上的明显未解析JIRA：|137338|137339|SPARK-7549:+Support+aggregating+over+nested+fields|137340|137341|entityMap|0|LINK|mutability|MUTABLE|url|https://issues.apache.org/jira/browse/SPARK-13721|1|https://issues.apache.org/jira/browse/SPARK-7549^0|0|0|0|0|0|O|I|1V|5|0|0|0|1L|0|0|0|0|0|0|1E|1|0|0^^$0|@$1|2|3|4|5|6|7|1D|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|1E|8|@]|9|@]|A|$]]|$1|D|3|E|5|F|7|1F|8|@]|9|@]|A|$G|H]]|$1|I|3|J|5|6|7|1G|8|@]|9|@]|A|$]]|$1|K|3|L|5|6|7|1H|8|@]|9|@]|A|$]]|$1|M|3|N|5|6|7|1I|8|@$O|1J|P|1K|Q|R]|$O|1L|P|1M|Q|R]]|9|@]|A|$]]|$1|S|3|-4|5|6|7|1N|8|@]|9|@]|A|$]]|$1|T|3|U|5|V|7|1O|8|@]|9|@$O|1P|P|1Q|1|1R]]|A|$]]|$1|W|3|-4|5|6|7|1S|8|@]|9|@]|A|$]]|$1|X|3|Y|5|6|7|1T|8|@]|9|@]|A|$]]|$1|Z|3|-4|5|6|7|1U|8|@]|9|@]|A|$]]|$1|10|3|11|5|V|7|1V|8|@]|9|@$O|1W|P|1X|1|1Y]]|A|$]]|$1|12|3|-4|5|6|7|1Z|8|@]|9|@]|A|$]]|$1|13|3|-4|5|6|7|20|8|@]|9|@]|A|$]]]|14|$15|$5|16|17|18|A|$19|1A]]|1B|$5|16|17|18|A|$19|1C]]]]

Above answers are all great answers and tackle this question from different sides; Spark SQL is also quite useful way to access nested data.

Here's example how to use explode() in SQL directly to query nested collection.

<pre><code>SELECT hholdid, tsp.person_seq_no 
FROM ( SELECT hholdid, explode(tsp_ids) as tsp 
 FROM disc_mrt.unified_fact uf
 )
</code></pre>

tsp_ids is a nested of structs, which has many attributes, including person_seq_no which I'm selecting in the outer query above.

Above was tested in Spark 2.0. I did a small test and it doesn't work in Spark 1.6. This question was asked when Spark 2 wasn't around, so this answer adds nicely to the list of available options to deal with nested structures.

Have a look also on following JIRAs for Hive-compatible way to query nested data using <code>LATERAL VIEW OUTER</code> syntax, since Spark 2.2 also supports <code>OUTER</code> explode (e.g. when a nested collection is empty, but you still want to have attributes from a parent record):

<ul>
<li><a href="https://issues.apache.org/jira/browse/SPARK-13721" rel="nofollow noreferrer">SPARK-13721: Add support for LATERAL VIEW OUTER explode()</a></li>
</ul>

Noticable not resolved JIRA on explode() for SQL access:

<ul>
<li><a href="https://issues.apache.org/jira/browse/SPARK-7549" rel="nofollow noreferrer">SPARK-7549: Support aggregating over nested fields</a></li>
</ul>

I have a parquet table with one of the columns being 

<blockquote>
 , array&lt;struct&lt;col1,col2,..colN>>
</blockquote>

Can run queries against this table in Hive using LATERAL VIEW syntax.

How to read this table into an RDD, and more importantly how to filter, map etc this nested collection in Spark?

Could not find any references to this in Spark documentation. Thanks in advance for any information!

ps. I felt might be helpful to give some stats on the table. 
 Number of columns in main table ~600. Number of rows ~200m.
 Number of "columns" in nested collection ~10. Avg number of records in nested collection ~35.

How to read a nested collection in Spark

Python

Spark 

Hive

我有一张镶木镶木桌上有一根柱子、array<struct<col1、col2、..colN>>可以使用横向视图语法在配置单元中对此表运行查询。如何将该表读入RDD，更重要的是如何过滤、映射等Spark中的嵌套集合？在Spark文档中找不到任何与此相关的引用。提前感谢您提供的任何信息！ps。我觉得在谈判桌上提供一些统计数...

问如何读取Spark中的嵌套集合
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如何读取Spark中的嵌套集合EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如何读取Spark中的嵌套集合
EN