PySpark中RDD与DataFrame相互转换操作

迷南。 2022-04-24 07:34 881阅读 0赞

### 1. 弹性数据集RDD ###

RDD是一个抽象的分布式数据集合，它提供了一系列转化操作（例如基本的map()、flatMap()、filter()，类集合操作union()、intersection()、subtract()）和行动操作（例如collect()、count()、take()、top()、reduce()、foreach()）。可以说，RDD是非常灵活的数据集合，其中可以存放类型相同或者互异的数据，同时可以指定任何自己期望的函数对其中的数据进行处理。   
创建一个RDD：

> \# 从list中创建  
> rdd = sc.parallelize(\[1, '2', (3, 4), \['5', '6'\]\])  
> \# 从文件中读取  
> rdd = sc.textFile('\\path\\to\\file')

还有一类RDD是key-value Pair RDD，即规定RDD每个元素都是一个二元组，其中第一个值是key，第二个值为value，key一般选取RDD中每个元素的一个字段。   
创建一个Pair RDD：

> \# 创建一个普通RDD  
> rdd = sc.parallelize(\[('a', 1, 2), ('b', 3, 4), ('c', 5, 6)\])  
> \# 提取每个元素的第一个元素作为key剩余元素作为value创建Pair RDD  
> pair\_rdd = rdd.map(lambda x: (x\[0\], x\[1:\]))

可以看到Pair RDD实质上仍然是一个普通的RDD，那为什么它要单独拿出来讲呢？   
这是因为，Pair RDD由于有key的存在，与普通的RDD相比更加格式化，这种特性就会给Pair RDD赋予一些特殊的操作，例如groupByKey()可以将具有相同key进行分组，其结果仍然得到Pair RDD，然后利用mapValues()对相同key的value进行函数计算；reduceByKey()、countByKey()和sortByKey()等一系列“ByKey()”操作同理。   
另外，两个Pair RDD具有像SQL一样的连接操作，例如两个Pair RDD进行join()后，具有相同key的元素的value会被放在一个元组里，key不相同的元素会被舍弃。leftOuterJoin()、rightOuterJoin()、fullOuterJoin()等操作同理。

### 2. Spark SQL中的DataFrame ###

Pair RDD已经被一定程度的格式化了，它的每个元素会具有key，但是value仍然具有很大的灵活性。DataFrame是一种完全格式化的数据集合，和数据库中的表的概念比较接近，它每列数据必须具有相同的数据类型。也正是由于DataFrame知道数据集合所有的类型信息，DataFrame可以进行列处理优化而获得比RDD更优的性能。   
在内部实现上，DataFrame是由Row对象为元素组成的集合，每个Row对象存储DataFrame的一行，Row对象中记录每个域=>值的映射，因而Row可以被看做是一个结构体类型。可以通过创建多个tuple/list、dict、Row然后构建DataFrame。

> >>> dicts = \[\{'col1':'a', 'col2':1\}, \{'col1':'b', 'col2':2\}\]  
> >>> dict\_dataframe = sqlContext.createDataFrame(dicts)  
> >>> dict\_dataframe.show()  
> \+----+----+  
> |col1|col2|  
> \+----+----+  
> |   a|   1|  
> |   b|   2|  
> \+----+----+
> 
> >>> lists = \[\['a', 1\], \['b', 2\]\]  
> >>> list\_dataframe = sqlContext.createDataFrame(lists,\['col1','col2'\])  
> >>> list\_dataframe.show()  
> \+----+----+                                                                       
> |col1|col2|  
> \+----+----+  
> |   a|   1|  
> |   b|   2|  
> \+----+----+
> 
> >>> from pyspark.sql import Row  
> >>> rows = \[Row(col1='a', col2=1), Row(col1='b', col2=2)\]  
> >>> row\_dataframe= sqlContext.createDataFrame(rows)  
> >>> row\_dataframe.show()  
> \+----+----+  
> |col1|col2|  
> \+----+----+  
> |   a|   1|  
> |   b|   2|  
> \+----+----+

虽然DataFrame被完全格式化了，但是其中每列可以存储的类型仍然是非常丰富的，包括基本的数据类型、list、tuple、dict和Row，这也就意味着所有的复杂数据类型都可以相互嵌套，从而解除了完全格式化的限制。例如，你可以在一列中存储list类型，而每行list按需存储不定长的数据。   
那么，RDD和DataFrame还有哪些使用上的区别呢？

RDD：没有列名称，只能使用数字来索引；具有map()、reduce()等方法并可指定任意函数进行计算;  
DataFrame：一定有列名称（即使是默认生成的），可以通过.col\_name或者\['col\_name'\]来索引列；具有表的相关操作（例如select()、filter()、where()、join），但是没有map()、reduce()等方法。

### 3. RDD转换为DataFrame ###

什么样的RDD可以转换为DataFrame？   
RDD灵活性很大，并不是所有RDD都能转换为DataFrame，而那些每个元素具有一定相似格式的时候才可以。

为什么RDD需要转换为DataFrame？   
当RDD进行类似表的相应操作时，都需要指定相应的函数，转换为DataFrame书写更简单，并且执行效率高。

怎么样将RDD转换为DataFrame？   
就像之前的例子一样，可以利用  
`sqlContext.createDataFrame`(*data*, *schema=None*, *samplingRatio=None*, *verifySchema=True*)

来将RDD转换为DataFrame，其中的参数设置需要注意：   
schema：DataFrame各列类型信息，在提前知道RDD所有类型信息时设定。例如

schema = StructType(\[StructField('col1', StringType()), StructField('col2', IntegerType())\])  
  
samplingRatio：推测各列类型信息的采样比例，在未知RDD所有类型信息时，spark需要根据一定的数据量进行类型推测；默认情况下，spark会抽取前100的RDD进行推测，之后在真正将RDD转换为DataFrame时如果遇到类型信息不符会报错 Some of types cannot be determined by the first 100 rows, please try again with sampling 。同理采样比例较低，推测类型信息也可能错误。

>>> from pyspark.sql.types import *
    >>> schema = StructType([StructField('col1', StringType()), StructField('col2', IntegerType())])
    >>> rdd=sc.parallelize([('key1',1111),('key2',2222),('key3',3333)])
    >>> pp11=sqlContext.createDataFrame(rdd)
    >>> pp11.show()
    +----+----+
    |  _1|  _2|
    +----+----+
    |key1|1111|
    |key2|2222|
    |key3|3333|
    +----+----+
    
    >>> pp22=sqlContext.createDataFrame(rdd,schema)
    >>> pp22.show()
    +----+----+
    |col1|col2|
    +----+----+
    |key1|1111|
    |key2|2222|
    |key3|3333|
    +----+----+

参考:[https://stackoverflow.com/questions/30905515/pyspark-structtype-is-not-defined][https_stackoverflow.com_questions_30905515_pyspark-structtype-is-not-defined]

4. DataFrame转换为RDD

有时候DataFrame的表相关操作不能处理一些问题，例如需要对一些数据利用指定的函数进行计算时，就需要将DataFrame转换为RDD。DataFrame可以直接利用.rdd获取对应的RDD对象，此RDD对象的每个元素使用Row对象来表示，每列值会成为Row对象的一个域=>值映射。例如

> >>> lists = \[\['a', 1\], \['b', 2\]\]  
> >>> list\_dataframe = sqlContext.createDataFrame(lists,\['col1','col2'\])  
> >>> list\_dataframe.show()  
> \+----+----+                                                                       
> |col1|col2|  
> \+----+----+  
> |   a|   1|  
> |   b|   2|  
> \+----+----+
> 
> >>> rdd=list\_dataframe.rdd  
> >>> rdd.collect()  
> \[Row(col1=u'a', col2=1), Row(col1=u'b', col2=2)\] 
> 
> >>> rdd.map(lambda x: \[x\[0\], x\[1:\]\]).collect()  
> \[\[u'a', (1,)\], \[u'b', (2,)\]\]  
> >>> rdd.map(lambda x: \[x\[0\], x\[1\]\]).collect()  
> \[\[u'a', 1\], \[u'b', 2\]\]

DataFrame转化后的RDD如果需要和一般形式的RDD进行操作（例如join），还需要做索引将数值从Row中取出，比如转化为Pair RDD可以这样操作

> >>> rdd.map(lambda x: \[str(x\[0\]), x\[1\]\]).collect()  
> \[\['a', 1\], \['b', 2\]\]  
> >>> rdd.map(lambda x: (str(x\[0\]), x\[1\])).collect()  
> \[('a', 1), ('b', 2)\]

注意：DataFrame转化的RDD可能包含Row(col1='a')，它和'a'是不同的对象，所以如果与一般的RDD进行join，还需要索引Row取出数值。

参考：[http://spark.apache.org/docs/latest/api/python/pyspark.sql.html][http_spark.apache.org_docs_latest_api_python_pyspark.sql.html]  
          https://blog.csdn.net/zhufenghao/article/details/80712480

[https_stackoverflow.com_questions_30905515_pyspark-structtype-is-not-defined]: https://stackoverflow.com/questions/30905515/pyspark-structtype-is-not-defined
[http_spark.apache.org_docs_latest_api_python_pyspark.sql.html]: http://spark.apache.org/docs/latest/api/python/pyspark.sql.html