pyspark之RDD

RDD簡介

RDD就是一個不可變的分布式對象集合—彈性分布式資料集（Resilient Distributed Dataset）。每個RDD都被分為多個分區，這些分區運作在叢集中的不同節點上。RDD可以包含任意類型的對象，包括使用者自定義的對象。

RDD常用操作

建立RDD

1.對一個集合進行并行化

x = sc.parallelize([1.0, 0.0, -2.0], 2)
           

2.讀取外部資料集

textFile = sc.textFile(path)
           

轉化操作(轉化操作會形成新的RDD)

針對各個元素的轉化操作

兩個最常用的轉化操作是map()和filter()。轉化操作map()接收一個函數，将這個函數用于RDD中的每個元素，将函數的傳回結果作為結果RDD中對應元素的值。轉化操作filter()接收一個函數，并将RDD中滿足該函數的元素放入新的RDD中傳回。

1.map()

df1=df.select('cpurate','memrate','ser','plr','temp','alldevicenumber','opticalinpower','opticaloutpower','dialingnumber')
df1.show(5)

+-------+-------+---+---+----+---------------+--------------+---------------+-------------+
|cpurate|memrate|ser|plr|temp|alldevicenumber|opticalinpower|opticaloutpower|dialingnumber|
+-------+-------+---+---+----+---------------+--------------+---------------+-------------+
|    2.0|   83.0|0.0|0.0|52.0|              1|         -21.2|            2.0|            0|
|    1.0|   72.0|0.0|0.0|49.0|              2|         -17.0|            2.0|            0|
|    1.0|   68.0|0.0|0.0|37.0|              0|         -22.7|            2.1|            0|
|    1.0|   47.0|0.0|0.0|41.0|              1|         -18.3|            2.2|            0|
|    1.0|   69.0|0.0|0.0|41.0|              1|         -17.5|            2.4|            3|
+-------+-------+---+---+----+---------------+--------------+---------------+-------------+
only showing top 5 rows

#DataFrame.rdd是将DataFrame轉化為一個rdd,rdd中的每個元素是DataFrame中的Row
df_rdd1 = df1.rdd
df_rdd1.take(5)
[Row(cpurate=2.0, memrate=83.0, ser=0.0, plr=0.0, temp=52.0, alldevicenumber=1, opticalinpower=-21.200000762939453, opticaloutpower=2.0, dialingnumber=0), Row(cpurate=1.0, memrate=72.0, ser=0.0, plr=0.0, temp=49.0, alldevicenumber=2, opticalinpower=-17.0, opticaloutpower=2.0, dialingnumber=0), Row(cpurate=1.0, memrate=68.0, ser=0.0, plr=0.0, temp=37.0, alldevicenumber=0, opticalinpower=-22.700000762939453, opticaloutpower=2.0999999046325684, dialingnumber=0), Row(cpurate=1.0, memrate=47.0, ser=0.0, plr=0.0, temp=41.0, alldevicenumber=1, opticalinpower=-18.299999237060547, opticaloutpower=2.200000047683716, dialingnumber=0), Row(cpurate=1.0, memrate=69.0, ser=0.0, plr=0.0, temp=41.0, alldevicenumber=1, opticalinpower=-17.5, opticaloutpower=2.4000000953674316, dialingnumber=3)]

#對RDD(df1.rdd)中的每個元素(df1中的一個Row)提取數值，組成新的元素
df_rdd = df1.rdd.map(lambda row: row[0:])
df_rdd.take(5)
[(2.0, 83.0, 0.0, 0.0, 52.0, 1, -21.200000762939453, 2.0, 0), (1.0, 72.0, 0.0, 0.0, 49.0, 2, -17.0, 2.0, 0), (1.0, 68.0, 0.0, 0.0, 37.0, 0, -22.700000762939453, 2.0999999046325684, 0), (1.0, 47.0, 0.0, 0.0, 41.0, 1, -18.299999237060547, 2.200000047683716, 0), (1.0, 69.0, 0.0, 0.0, 41.0, 1, -17.5, 2.4000000953674316, 3)]
           

行動操作

行動操作有collect()[傳回RDD中的所有元素]、take(n)[從RDD中傳回n個元素]等