弹性分布式数据集（RDD）不仅是一组不可变的JVM对象的分布集，可以执行高速运算，而且是Apache Spark的核心。该数据集基于关键字将数据集划分成块，同时分发到执行器节点。这样做可以使此类数据集能够高速执行运算。另外，RDD将跟踪应用于每个块的转换，以加快计算速度。
1.RDD运行方式
1.1创建RDD
Pyspark中，有两种方式创建RDD
1 .用parallelize()创建
data = sc.parallelize([('Amber',22),('Alfred',23),('Skye',4),('Albert',12),('Amber',9)])
2.读取文件方式创建
data_from_file = sc.textFile('VS14MORT.DUSMCPUB',4)
sc.textFile()支持多种数据格式：文本、parquet、JSON、Hive table以及使用JDBC驱动程序可读取的关系数据库中的数据。
1.2 Lambda表达式
通过Lambda将不可读的行解析成我们能够使用的信息。
def extractInformation(row):
import re
import numpy as np
selected_indices = [
2,4,5,6,7,9,10,11,12,13,14,15,16,17,18,
19,21,22,23,24,25,27,28,29,30,32,33,34,
36,37,38,39,40,41,42,43,44,45,46,47,48,
49,50,51,52,53,54,55,56,58,60,61,62,63,
64,65,66,67,68,69,70,71,72,73,74,75,76,
77,78,79,81,82,83,84,85,87,89
]
record_split = re\
.compile(
r'([\s]{19})([0-9]{1})([\s]{40})([0-9\s]{2})([0-9\s]{1})([0-9]{1})([0-9]{2})' +
r'([\s]{2})([FM]{1})([0-9]{1})([0-9]{3})([0-9\s]{1})([0-9]{2})([0-9]{2})' +
r'([0-9]{2})([0-9\s]{2})([0-9]{1})([SMWDU]{1})([0-9]{1})([\s]{16})([0-9]{4})' +
r'([YNU]{1})([0-9\s]{1})([BCOU]{1})([YNU]{1})([\s]{34})([0-9\s]{1})([0-9\s]{1})' +
r'([A-Z0-9\s]{4})([0-9]{3})([\s]{1})([0-9\s]{3})([0-9\s]{3})([0-9\s]{2})([\s]{1})' +
r'([0-9\s]{2})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})' +
r'([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})' +
r'([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})' +
r'([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})([A-Z0-9\s]{7})' +
r'([A-Z0-9\s]{7})([\s]{36})([A-Z0-9\s]{2})([\s]{1})([A-Z0-9\s]{5})([A-Z0-9\s]{5})' +
r'([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})' +
r'([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})' +
r'([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})' +
r'([A-Z0-9\s]{5})([A-Z0-9\s]{5})([A-Z0-9\s]{5})([\s]{1})([0-9\s]{2})([0-9\s]{1})' +
r'([0-9\s]{1})([0-9\s]{1})([0-9\s]{1})([\s]{33})([0-9\s]{3})([0-9\s]{1})([0-9\s]{1})')
try:
rs = np.array(record_split.split(row))[selected_indices]
except:
rs = np.array(['-99'] * len(selected_indices))
return rs
data_from_file_conv = data_from_file.map(extractInformation)
纯python方法会降低应用程序的速度，因为Spark需要再python解释器和JVM之间连续切换。所以应该尽可能使用内置的Spark功能。
1.3 转换
转换可以调整数据集。包括映射、筛选、连接、转换数据集中的值。
1.3.1 map转换
data_2014 = data_from_file_conv.map(lambda row: int(row[16]))
data_2014.take(10)
[out]: [2014, 2014, 2014, 2014, 2014, 2014, 2014, 2014, 2014, -99]
1.3.2 filter转换
filter方法可以从数据集中选择元素，该元素集符合特定的标准。比如统计2014年车祸死亡人数。
data_filtered = data_from_file_conv.filter(lambda row: row[5] == 'F' and row[21] == '0')
data_filtered.count()
1.3.3 flatmap转换
flatmap方法和map方法类似，但是flatmap返回的是一个扁平的结果，而不是一个列表。
data_2014_flat = data_from_file_conv.flatMap(lambda row: (row[16], int(row[16]) + 1))
data_2014_flat.take(10)
[out]: ['2014', 2015, '2014', 2015, '2014', 2015, '2014', 2015, '2014', 2015]
1.3.4 repartition转换
repartition可以对数据集重新分区，改变数据集分区数量，此功能应该谨慎并且仅当真正需要的时候再使用，因为它会重组数据，导致对性能方面产生巨大影响。
rdd1 = rdd.repartition(4)
1.4 操作
和转换不同，操作执行数据集上的计划任务。一旦完成数据转换，则可以执行相应转换。操作包括take、collect、reduce、count、saveAsTextFile、foreach等。这些应该都是再driver端执行。