PySpark 学习笔记一

淩亂°似流年 2022-06-02 12:09 324阅读 0赞

为了更好地进行大数据分析与处理，最近在学习PySpark，整理了一下笔记 ，加深印象。

**1 Resilient Distributed Datasets（RDD）**

弹性分布式数据集（RDD）是一个不可变的JVM对象的分布式集合，是Spark的基本抽象。

**1.1 创建RDD**  
准备工作：

>>> import pyspark
    >>> from pyspark import SparkContext
    >>> from pyspark import SparkConf
    >>> conf = SparkConf().setAppName('project1').setMaster('local')
    >>> sc = SparkContext.getOrCreate(conf)

在PySpark里有两种方法创建RDD:  
一是，.parallelize(…) 个collection集合 ( list or an array of some elements)。

>>> data = sc.parallelize([('amber',22),('alfred',23),('skye',4),('albert',12),('amber',9)])

二是，引用位于本地或HDFS上的某个文件（或多个文件）。

>>> data_from_file = sc.textFile('/home/qml/pyspark-ex/VS14MORT.txt.gz',4)
    # sc.textFile（...，n）中的最后一个参数指定数据集被分区的数量，经验是分成两个四分区

Spark 支持多种数据格式：可以使用JDBC驱动程序读取文本，Parquet，JSON，Hive表和来自关系数据库的数据。请注意，Spark可以自动处理压缩的数据集（如Gzip压缩数据集）。  
从文件读取的数据表示为MapPartitionsRDD，而不是像当我们.paralellize（…）一个集合的数据一样表示为ParallelCollectionRDD。

**1.2 Schema**

RDD是无模式的数据结构（不像DataFrames）。因此，在使用RDD时，并行化数据集对于Spark来说是完美的。

>>> data_heterogenous = sc.parallelize([('Ferrari','fast'),{
       'Porsche':100000},['Spain','visited',4504]]).collect()

所以，我们可以混合几乎任何东西：一个元组，一个字典，或一个列表。  
一旦你.collect（）数据集（即，运行一个动作将其返回给驱动程序），你可以像在Python中通常那样访问对象中的数据：

>>> data_heterogenous[1]['Porsche']
    100000

.collect（）方法将RDD的所有元素返回到驱动程序，并将其作为列表序列化。

**1.3 读取文件**

从文本文件读取时，文件中的每一行形成RDD的一个元素。 可以创建一个元素列表，每行代表一个值列表。

>>> data_from_file.take(1)
    
    [u'                   1                                          2101  M1087 432311  4M4                2014U7CN                                    I64 238 070   24 0111I64                                                                                                                                                                           01 I64                                                                                                  01  11                                 100 601']

**1.4 Lambda表达式**  
**1.4.1 Transformations**

*  **.map(…)**

该方法应用于RDD的每个元素。

In [1]:
    data_2014_2 = data_from_file_conv.map(lambda row: (row[16], int(row[16])))
    data_2014_2.take(10)
    
    Out[2]:
    [('2014', 2014),
     ('2014', 2014),
     ('2014', 2014),
     ('2014', 2014),
     ('2014', 2014),
     ('2014', 2014),
     ('2014', 2014),
     ('2014', 2014),
     ('2014', 2014),
     ('-99', -99)]

*  **.filter(…)**

允许选择符合指定条件的数据集元素。

data_filtered = data_from_file_conv.filter(lambda row: row[5] == 'F' and row[21] == '0')
    data_filtered.count()

*  **.flatMap(…)**  
    与map()的工作方式类似，但返回的是平铺的结果而不是列表。

In [3]:
    data_2014_flat = data_from_file_conv.flatMap(lambda row: (row[16], int(row[16]) + 1))
    data_2014_flat.take(10)
    
    Out[4]:
    ['2014', 2015, '2014', 2015, '2014', 2015, '2014', 2015, '2014', 2015]

*  **.distinct()**

此方法返回指定列中不同值的列表。

In [5]:
    distinct_gender = data_from_file_conv.map(lambda row: row[5]).distinct().collect()
    distinct_gender
    
    Out[6]:
    ['-99', 'M', 'F']

*  **.sample(…)**

该方法返回数据集中的随机样本。

In [7]:
    fraction = 0.1
    data_sample = data_from_file_conv.sample(False, fraction, 666)
    
    data_sample.take(1)
    
    Out[8]:
    [array(['1', '  ', '5', '1', '01', 'F', '1', '082', ' ', '42', '22', '10',
            '  ', '4', 'W', '5', '2014', 'U', '7', 'C', 'N', ' ', ' ', 'I251',
            '215', '063', '   ', '21', '02', '11I350 ', '21I251 ', '       ',
            '       ', '       ', '       ', '       ', '       ', '       ',
            '       ', '       ', '       ', '       ', '       ', '       ',
            '       ', '       ', '       ', '       ', '       ', '02',
            'I251 ', 'I350 ', '     ', '     ', '     ', '     ', '     ',
            '     ', '     ', '     ', '     ', '     ', '     ', '     ',
            '     ', '     ', '     ', '     ', '     ', '     ', '28', ' ',
            ' ', '2', '4', '100', '8'], 
           dtype='<U40')]

*  **.leftOuterJoin(…)**

左外连接就像SQL一样，根据两个数据集中的值加入两个RDD，并从左RDD中返回从右侧追加两个RDD匹配的记录。

>>> rd1 = sc.parallelize([('a',1),('b',4),('c',10)])
    >>> rd2 = sc.parallelize([('a',4),('a',1),('b','6'),('d',15)])
    >>> rd3 = rd1.leftOuterJoin(rd2)
    >>> print rd3.take(5)
    [('a', (1, 4)), ('a', (1, 1)), ('c', (10, None)), ('b', (4, '6'))]

如果我们使用.join（…）方法，那么当这两个值在这两个RDD之间相交时，我们只能得到’a’和’b’的值。

>>> rd4 = rd1.join(rd2)
    >>> print rd4.collect()
    [('a', (1, 4)), ('a', (1, 1)), ('b', (4, '6'))]

另一个有用的方法是.intersection（…），它返回两个RDD中相同的记录。

>>> rd5 = rd1.intersection(rd2)
    >>> print rd5.collect()
    [('a', 1)]

**1.4.2 Actions**

*  **.take(…)**

该方法从单个数据分区返回n个最高行。

In [9]:
    data_first = data_from_file_conv.take(1)
    data_first
    
    Out[10]:
    [array(['1', '  ', '2', '1', '01', 'M', '1', '087', ' ', '43', '23', '11',
            '  ', '4', 'M', '4', '2014', 'U', '7', 'C', 'N', ' ', ' ', 'I64 ',
            '238', '070', '   ', '24', '01', '11I64  ', '       ', '       ',
            '       ', '       ', '       ', '       ', '       ', '       ',
            '       ', '       ', '       ', '       ', '       ', '       ',
            '       ', '       ', '       ', '       ', '       ', '01',
            'I64  ', '     ', '     ', '     ', '     ', '     ', '     ',
            '     ', '     ', '     ', '     ', '     ', '     ', '     ',
            '     ', '     ', '     ', '     ', '     ', '     ', '01', ' ',
            ' ', '1', '1', '100', '6'], 
           dtype='<U40')]

*  **.reduce(…)**

>>> rd1.map(lambda row: row[1]).reduce(lambda x,y:x+y)
    15

*  **.reduceByKey(…)**

>>> data_key = sc.parallelize([('a', 4),('b', 3),('c', 2),('a', 8),('d', 2),('b', 1),('d', 3)],4)
    >>> data_key.reduceByKey(lambda x, y: x + y).collect()
    [('b', 4), ('c', 2), ('a', 12), ('d', 5)]

*  **.count()**

统计RDD中元素的数量。

>>> data_reduce.count()
    6

*  **.countByKey()**

如果你的数据集是键值的形式，则可以使用.countByKey（）方法获取不同键的数量。

>>> data_key.countByKey().items()
    [('a', 2), ('b', 2), ('d', 2), ('c', 1)]

*  **.saveAsTextFile(…)**

将RDD保存到文本文件：每个分区保存到一个单独的文件。

>>> data_key.saveAsTextFile('/Users/drabast/Documents/PySpark_Data/data_key.txt')

*  **.foreach(…)**

一种将函数应用同到RDD每个元素的迭代法。

def f(x): 
        print(x)
    
    data_key.foreach(f)