美文网首页Spark学习笔记
12.Spark学习(Python版本):DStream转换操作

12.Spark学习(Python版本):DStream转换操作

作者: 马淑 | 来源:发表于2018-09-01 12:54 被阅读91次
    DStream转换操作包括无状态转换和有状态转换。

    无状态转换:每个批次的处理不依赖于之前批次的数据。
    有状态转换:当前批次的处理需要使用之前批次的数据或者中间结果。有状态转换包括基于滑动窗口的转换和追踪状态变化的转换(updateStateByKey)。

    DStream无状态转换操作
    • map(func) :对源DStream的每个元素,采用func函数进行转换,得到一个新的DStream;
    • flatMap(func): 与map相似,但是每个输入项可用被映射为0个或者多个输出项;
    • filter(func): 返回一个新的DStream,仅包含源DStream中满足函数func的项;
    • repartition(numPartitions): 通过创建更多或者更少的分区改变DStream的并行程度;
    • union(otherStream): 返回一个新的DStream,包含源DStream和其他DStream的元素;
    • count():统计源DStream中每个RDD的元素数量;
    • reduce(func):利用函数func聚集源DStream中每个RDD的元素,返回一个包含单元素RDDs的新DStream;
      *countByValue():应用于元素类型为K的DStream上,返回一个(K,V)键值对类型的新DStream,每个键的值是在原DStream的每个RDD中的出现次数;
    • reduceByKey(func, [numTasks]):当在一个由(K,V)键值对组成的DStream上执行该操作时,返回一个新的由(K,V)键值对组成的DStream,每一个key的值均由给定的recuce函数(func)聚集起来;
    • join(otherStream, [numTasks]):当应用于两个DStream(一个包含(K,V)键值对,一个包含(K,W)键值对),返回一个包含(K, (V, W))键值对的新DStream;
    • cogroup(otherStream, [numTasks]):当应用于两个DStream(一个包含(K,V)键值对,一个包含(K,W)键值对),返回一个包含(K, Seq[V], Seq[W])的元组;
    • transform(func):通过对源DStream的每个RDD应用RDD-to-RDD函数,创建一个新的DStream。支持在新的DStream中做任何RDD操作。
    DStream有状态转换操作

    对于DStream有状态转换操作而言,当前批次的处理需要使用之前批次的数据或者中间结果。
    有状态转换包括基于滑动窗口的转换和追踪状态变化(updateStateByKey)的转换。

    滑动窗口转换操作

    滑动窗口转换操作的计算过程如下图所示,我们可以事先设定一个滑动窗口的长度(也就是窗口的持续时间),并且设定滑动窗口的时间间隔(每隔多长时间执行一次计算),然后,就可以让窗口按照指定时间间隔在源DStream上滑动,每次窗口停放的位置上,都会有一部分DStream被框入窗口内,形成一个小段的DStream,这时,就可以启动对这个小段DStream的计算。


    • window(windowLength, slideInterval) 基于源DStream产生的窗口化的批数据,计算得到一个新的DStream;
    • countByWindow(windowLength, slideInterval) 返回流中元素的一个滑动窗口数;
    • reduceByWindow(func, windowLength, slideInterval) 返回一个单元素流。利用函数func聚集滑动时间间隔的流的元素创建这个单元素流。函数func必须满足结合律,从而可以支持并行计算;
    • reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])应用到一个(K,V)键值对组成的DStream上时,会返回一个由(K,V)键值对组成的新的DStream。每一个key的值均由给定的reduce函数(func函数)进行聚合计算。注意:在默认情况下,这个算子利用了Spark默认的并发任务数去分组。可以通过numTasks参数的设置来指定不同的任务数;
    • reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])更加高效的reduceByKeyAndWindow,每个窗口的reduce值,是基于先前窗口的reduce值进行增量计算得到的;它会对进入滑动窗口的新数据进行reduce操作,并对离开窗口的老数据进行“逆向reduce”操作。但是,只能用于“可逆reduce函数”,即那些reduce函数都有一个对应的“逆向reduce函数”(以InvFunc参数传入);
    • countByValueAndWindow(windowLength, slideInterval, [numTasks]) 当应用到一个(K,V)键值对组成的DStream上,返回一个由(K,V)键值对组成的新的DStream。每个key的值都是它们在滑动窗口中出现的频率。
    updateStateByKey操作

    当我们需要在跨批次之间维护状态时,就必须使用updateStateByKey操作。
    在之前的套接字流的介绍中,我们统计单词词频采用的是无状态转换操作,NetworkWordCount只对本批次内的单词进行词频统计,不同批次的单词词频都是独立统计的。
    对于有状态转换操作而言,本批次的词频统计,会在之前批次的词频统计结果的基础上进行不断累加,所以,最终统计得到的词频,是所有批次的单词的总的词频统计结果。
    请登录Linux系统,打开一个终端,然后,执行下面命令:

    cd /usr/local/spark/mycode/streaming 
    vim NetworkWordCountStateful.py
    

    NetworkWordCountStateful.py中输入以下代码:

    from __future__ import print_function
     
    import sys
     
    from pyspark import SparkContext
    from pyspark.streaming import StreamingContext
     
    if __name__ == "__main__":
        if len(sys.argv) != 3:
            print("Usage: stateful_network_wordcount.py <hostname> <port>", file=sys.stderr)
            exit(-1)
        sc = SparkContext(appName="PythonStreamingStatefulNetworkWordCount")
        ssc = StreamingContext(sc, 1)
        ssc.checkpoint("file:///usr/local/spark/mycode/streaming/")
     
        # RDD with initial state (key, value) pairs
        initialStateRDD = sc.parallelize([(u'hello', 1), (u'world', 1)])
     
        def updateFunc(new_values, last_sum):
            return sum(new_values) + (last_sum or 0)
     
        lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2]))
        running_counts = lines.flatMap(lambda line: line.split(" "))\
                              .map(lambda word: (word, 1))\
                              .updateStateByKey(updateFunc, initialRDD=initialStateRDD)
     
        running_counts.pprint()
     
        ssc.start()
        ssc.awaitTermination()
    

    保存后退出vim编辑器,输入以下命令进入了监听状态(我们把运行这个监听程序的窗口称为监听窗口):
    python ./NetworkWordCountStateful.py localhost 9999

    新打开一个窗口作为nc窗口,启动nc程序:

    nc -lk 9999
    //在这个窗口中手动输入一些单词
    hadoop
    spark
    hadoop
    spark
    hadoop
    spark
    

    切换到刚才的监听窗口,会发现,已经输出了词频统计信息:


    相关文章

      网友评论

        本文标题:12.Spark学习(Python版本):DStream转换操作

        本文链接:https://www.haomeiwen.com/subject/ndzywftx.html