上面我們學(xué)習(xí)了RDD如何轉(zhuǎn)換，即一個(gè)RDD轉(zhuǎn)換成另外一個(gè)RDD，但是轉(zhuǎn)換完成之后并沒有立刻執(zhí)行，僅僅是記住了數(shù)據(jù)集的邏輯操作，只有當(dāng)執(zhí)行了Action動(dòng)作之后才會(huì)真正觸發(fā)Spark作業(yè)，進(jìn)行算子的計(jì)算

執(zhí)行操作有：

• reduce(func)
• collect()
• count()
• first()
• take(n)
• takeSample(withReplacement, num, [seed])
• takeOrdered(n, [ordering])
• saveAsTextFile(path)
• countByKey()
• foreach(func)

reduce：使用函數(shù)func聚合數(shù)據(jù)集元素，返回執(zhí)行結(jié)果

            
              from pyspark import SparkContext

if __name__ == '__main__':
    sc = SparkContext(appName="rddAction", master="local[*]")
    data = [1, 2, 3, 4, 5]
    rdd = sc.parallelize(data)
    print(rdd.reduce(lambda x,y : x+y))
    # 15
    sc.stop()
            
          

collect：將計(jì)算結(jié)果回收到Driver端，當(dāng)數(shù)據(jù)量較大時(shí)執(zhí)行會(huì)造成oom

            
              from pyspark import SparkContext

if __name__ == '__main__':
    sc = SparkContext(appName="rddAction", master="local[*]")
    data = [1, 2, 3, 4, 5]
    rdd = sc.parallelize(data)
    print(rdd.collect())
    # [1, 2, 3, 4, 5]
    sc.stop()
            
          

count：返回?cái)?shù)據(jù)集元素個(gè)數(shù)，執(zhí)行過程中會(huì)將數(shù)據(jù)回收到Driver端進(jìn)行統(tǒng)計(jì)

            
              from pyspark import SparkContext

if __name__ == '__main__':
    sc = SparkContext(appName="rddAction", master="local[*]")
    data = [1, 2, 3, 4, 5]
    rdd = sc.parallelize(data)
    print(rdd.count())
    # 5
    sc.stop()
            
          

first：返回?cái)?shù)據(jù)集中的第一個(gè)元素，類似于take(1)

            
              from pyspark import SparkContext

if __name__ == '__main__':
    sc = SparkContext(appName="rddAction", master="local[*]")
    data = [1, 2, 3, 4, 5]
    rdd = sc.parallelize(data)
    print(rdd.first())
    # 1
    sc.stop()
            
          

take：返回?cái)?shù)據(jù)集中的前n個(gè)元素的數(shù)組

            
              from pyspark import SparkContext

if __name__ == '__main__':
    sc = SparkContext(appName="rddAction", master="local[*]")
    data = [1, 2, 3, 4, 5]
    rdd = sc.parallelize(data)
    print(rdd.take(3))
    # [1, 2, 3]
    sc.stop()
            
          

takeSample：返回?cái)?shù)據(jù)集中num個(gè)隨機(jī)元素，seed指定隨機(jī)數(shù)生成器種子

            
              from pyspark import SparkContext

if __name__ == '__main__':
    sc = SparkContext(appName="rddAction", master="local[*]")
    data = [1, 2, 3, 4, 5]
    rdd = sc.parallelize(data)
    print(rdd.takeSample(True, 3, 1314))
    # [5, 2, 3]
    sc.stop()
            
          

takeOrdered：使用自然排序或自定義比較器返回?cái)?shù)據(jù)集中的前n個(gè)元素

            
              from pyspark import SparkContext

if __name__ == '__main__':
    sc = SparkContext(appName="rddAction", master="local[*]")
    data = [5, 1, 4, 2, 3]
    rdd = sc.parallelize(data)
    print(rdd.takeOrdered(3))
    # [1, 2, 3]
    print(rdd.takeOrdered(3, key=lambda x: -x))
    # [5, 4, 3]
    sc.stop()
            
          

saveAsTextFile：將數(shù)據(jù)集元素作為文本文件寫入文件系統(tǒng)（如：本地文件系統(tǒng)，HDFS等）

            
              from pyspark import SparkContext

if __name__ == '__main__':
    sc = SparkContext(appName="rddAction", master="local[*]")
    data = [1, 2, 3, 4, 5]
    rdd = sc.parallelize(data)
    rdd.saveAsTextFile("file:///home/data")
    sc.stop()
            
          

countByKey：統(tǒng)計(jì)(K,V)對(duì)中每個(gè)K的個(gè)數(shù)

            
              from pyspark import SparkContext

if __name__ == '__main__':
    sc = SparkContext(appName="rddAction", master="local[*]")
    data = [('a', 1), ('b', 2), ('a', 3)]
    rdd = sc.parallelize(data)
    print(sorted(rdd.countByKey().items()))
    # [('a', 2), ('b', 1)]
    sc.stop()
            
          

foreach：對(duì)RDD每個(gè)元素執(zhí)行指定函數(shù)

            
              from pyspark import SparkContext

def f(x):
    print(x)

if __name__ == '__main__':
    sc = SparkContext(appName="rddAction", master="local[*]")
    data = [1, 2, 3]
    rdd = sc.parallelize(data)
    rdd.foreach(f)
    # 1 2 3
    sc.stop()
            
          

至此，所有action動(dòng)作學(xué)習(xí)完畢

?

Spark學(xué)習(xí)目錄：

• Spark學(xué)習(xí)實(shí)例1(Python)：單詞統(tǒng)計(jì) Word Count
• Spark學(xué)習(xí)實(shí)例2(Python)：加載數(shù)據(jù)源Load Data Source
• Spark學(xué)習(xí)實(shí)例3(Python)：保存數(shù)據(jù)Save Data
• Spark學(xué)習(xí)實(shí)例4(Python)：RDD轉(zhuǎn)換 Transformations
• Spark學(xué)習(xí)實(shí)例5(Python)：RDD執(zhí)行 Actions
• Spark學(xué)習(xí)實(shí)例6(Python)：共享變量Shared Variables
• Spark學(xué)習(xí)實(shí)例7(Python)：RDD、DataFrame、DataSet相互轉(zhuǎn)換
• Spark學(xué)習(xí)實(shí)例8(Python)：輸入源實(shí)時(shí)處理 Input Sources Streaming
• Spark學(xué)習(xí)實(shí)例9(Python)：窗口操作 Window Operations