亚洲免费在线-亚洲免费在线播放-亚洲免费在线观看-亚洲免费在线观看视频-亚洲免费在线看-亚洲免费在线视频

歡迎來訪張軍網站:連續登錄【5】天博主自動為您充值【5】積分哦

Spark學習實例(Python):輸入源實時處理 Input Sources

系統 1545 0

在之前學習的RDD和DataFrame數據集主要處理的是離線數據,隨著時代發展進步,我們會發現越來越多數據是在源源不斷發回到數據中心,同時需要立刻響應給用戶,這樣的情況我們就會用到實時處理,常用的場景有實時顯示某商場一小時人流密度、實時顯示當天火車站人口總數等等。接下來從實時數據源說起,實時數據源主要有:

  • File Source
  • Socket Source
  • Flume Source
  • Kafka Source

File Source指的是文件作為數據來源,常用的有本地文件file和分布式系統hdfs,這邊以本地文件來說明,實現代碼如下

            
              from pyspark import SparkContext
from pyspark.streaming import StreamingContext

if __name__ == '__main__':
    sc = SparkContext(appName="inputSourceStreaming", master="local[*]")
    # 第二個參數指統計多長時間的數據
    ssc = StreamingContext(sc, 5)

    lines = ssc.textFileStream("file:///home/llh/data/streaming")
    counts = lines.flatMap(lambda line: line.split(" ")).map(lambda x: (x, 1)).reduceByKey(lambda a,b: a+b)
    counts.pprint()
    # -------------------------------------------
    # Time: 2019-07-31 18:11:55
    # -------------------------------------------
    # ('hong', 2)
    # ('zhang', 2)
    # ('li', 2)
    # ('san', 2)
    # ('wang', 2)
    ssc.start()
    ssc.awaitTermination()
            
          

然后不斷向/home/llh/data/streaming/目錄下拷貝文件,結果如上面注釋所示

Socket Source指網絡套接字作為數據來源,用命令nc模擬網絡發送信息,實現代碼如下

            
              from pyspark import SparkContext
from pyspark.streaming import StreamingContext

if __name__ == '__main__':
    sc = SparkContext(appName="inputSourceStreaming", master="local[*]")
    # 第二個參數指統計多長時間的數據
    ssc = StreamingContext(sc, 5)

    lines = ssc.socketTextStream("localhost", 9999)
    counts = lines.flatMap(lambda line: line.split(" ")).map(lambda x: (x, 1)).reduceByKey(lambda a,b: a+b)
    counts.pprint()
    # -------------------------------------------
    # Time: 2019-07-31 18: 43:25
    # -------------------------------------------
    # ('hadoop', 1)
    # ('spark', 1)
    ssc.start()
    ssc.awaitTermination()
            
          

命令端執行~$ nc -lk 9999

hadoop spark

之后運行代碼即可

Flume是一個高可用海量收集日志系統,因此可作為數據來源,實現代碼如下

            
              from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.flume import FlumeUtils

if __name__ == '__main__':
    sc = SparkContext(appName="inputSourceStreaming", master="local[*]")
    # 第二個參數指統計多長時間的數據
    ssc = StreamingContext(sc, 5)

    lines = FlumeUtils.createStream("localhost", 34545)
    counts = lines.flatMap(lambda line: line.split(" ")).map(lambda x: (x, 1)).reduceByKey(lambda a,b: a+b)
    counts.pprint()

    ssc.start()
    ssc.awaitTermination()
            
          

Kafka是一款分布式消息隊列,常作為中間件用于傳輸,隔離,Kafka是以上四種里面實際開發最常用的流式數據來源,一樣實現代碼如下

            
              from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

if __name__ == '__main__':
    sc = SparkContext(appName="inputSourceStreaming", master="local[*]")
    # 第二個參數指統計多長時間的數據
    ssc = StreamingContext(sc, 5)

    kvs = KafkaUtils.createDirectStream(ssc, "topic-name", "localhost:9092")
    lines = kvs.map(lambda x: x[1])
    counts = lines.flatMap(lambda line: line.split(" ")).map(lambda x: (x, 1)).reduceByKey(lambda a,b: a+b)
    counts.pprint()

    ssc.start()
    ssc.awaitTermination()
            
          

好了,以上就是實時處理主要數據來源,第四種最為重要必須掌握。

?

Spark學習目錄:

  • Spark學習實例1(Python):單詞統計 Word Count
  • Spark學習實例2(Python):加載數據源Load Data Source
  • Spark學習實例3(Python):保存數據Save Data
  • Spark學習實例4(Python):RDD轉換 Transformations
  • Spark學習實例5(Python):RDD執行 Actions
  • Spark學習實例6(Python):共享變量Shared Variables
  • Spark學習實例7(Python):RDD、DataFrame、DataSet相互轉換
  • Spark學習實例8(Python):輸入源實時處理 Input Sources Streaming
  • Spark學習實例9(Python):窗口操作 Window Operations

更多文章、技術交流、商務合作、聯系博主

微信掃碼或搜索:z360901061

微信掃一掃加我為好友

QQ號聯系: 360901061

您的支持是博主寫作最大的動力,如果您喜歡我的文章,感覺我的文章對您有幫助,請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧,狠狠點擊下面給點支持吧,站長非常感激您!手機微信長按不能支付解決辦法:請將微信支付二維碼保存到相冊,切換到微信,然后點擊微信右上角掃一掃功能,選擇支付二維碼完成支付。

【本文對您有幫助就好】

您的支持是博主寫作最大的動力,如果您喜歡我的文章,感覺我的文章對您有幫助,請用微信掃描上面二維碼支持博主2元、5元、10元、自定義金額等您想捐的金額吧,站長會非常 感謝您的哦!!!

發表我的評論
最新評論 總共0條評論
主站蜘蛛池模板: 亚洲第一二三四区 | 日本免费久久 | 亚洲人和日本人jzz护士 | 成人免费aa在线观看 | 亚洲成在人网站天堂一区二区 | 日日a.v拍夜夜添久久免费 | 亚洲国产精品久久日 | 国产成人综合网亚洲欧美在线 | 国产精品久久久久孕妇 | 色狠狠色综合久久8狠狠色 色狠狠婷婷97 | 久久久久视频精品网 | 久久久久久久久久鸭 | 99在线播放 | 亚洲精品一线观看 | 青青热在线精品视频免费 | 国产伦一区二区三区免费 | 福利色姬网站视频入口 | 久久黄色精品视频 | 久久99久久99精品免费看动漫 | 国产欧美一区二区三区精品 | 亚洲毛片一级带毛片基地 | 日日狠狠的日日日日 | 91色蝌蚪| 久久美女视频 | 日本一区高清 | 色综合桃花网 | 国产一区二区在线观看免费 | 久久精品国产精品亚洲人人 | 一级a毛片免费观看久久精品 | 日本一区色| 8090色 | 国产精品成人免费观看 | 久久久久久久国产精品毛片 | 色综综| 国产精品好好热在线观看 | 亚洲成a人v在线观看 | 欧美在线一区二区三区精品 | 免费看黄色的网站 | 中文字幕免费观看视频 | 毛片女女女女女女女女女 | 91精品国产视频 |