Spark Streaming編程初級實踐詳解

更新時間：2023年04月20日 09:31:38 作者：WHYBIGDATA

這篇文章主要為大家介紹了Spark Streaming編程初級實踐詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

寫在前面

Linux：CentOS7.5
Spark： spark-3.0.0-bin-hadoop3.2
Flume：Flume-1.9.0
IDE：IntelliJ IDEA2020.2.3

1. 安裝Flume

Flume是Cloudera提供的一個分布式、可靠、可用的系統(tǒng)，它能夠?qū)⒉煌瑪?shù)據(jù)源的海量日志數(shù)據(jù)進行高效收集、聚合、移動，最后存儲到一個中心化數(shù)據(jù)存儲系統(tǒng)中。Flume 的核心是把數(shù)據(jù)從數(shù)據(jù)源收集過來，再送到目的地。請到Flume官網(wǎng)下載Flume1.7.0安裝文件，下載地址如下：

www.apache.org/dyn/closer.…

或者也可以直接到本教程官網(wǎng)的“下載專區(qū)”中的“軟件”目錄中下載apache-flume-1.7.0-bin.tar.gz。

下載后，把Flume1.7.0安裝到Linux系統(tǒng)的“/usr/local/flume”目錄下，具體安裝和使用方法可以參考教程官網(wǎng)的“實驗指南”欄目中的“日志采集工具Flume的安裝與使用方法。

安裝命令

tar -zxvf apache-flume-1.9.0-bin.tar.gz -C /export/server/
mv apache-flume-1.9.0-bin/ flume-1.9.0
sudo vi /etc/profile
export FLUME_HOME=/usr/local/flume
export PATH=$PATH:$FLUME_HOME/bin
source /etc/profile
mv flume-env.sh.template flume-env.sh

查看版本號

bin/flume-ng version

2.使用Avro數(shù)據(jù)源測試Flume

題目描述

Avro可以發(fā)送一個給定的文件給Flume，Avro 源使用AVRO RPC機制。請對Flume的相關配置文件進行設置，從而可以實現(xiàn)如下功能：在一個終端中新建一個文件helloworld.txt（里面包含一行文本“Hello World”），在另外一個終端中啟動Flume以后，可以把helloworld.txt中的文本內(nèi)容顯示出來。

Flume配置文件

al.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.type = avro
a1.sources.r1.channels= c1
a1.sources.r1.bind = 0.0.0.0
al.sources.r1.port = 4141
a1.sinks.k1.type = logger
a1.channels.c1.type = memory
al.channels.c1.capacity = 1000
a1.channels.c1.transaction = 100
al.sources.r1.channels = c1
a1.sinks.k1.channel=c1

執(zhí)行命令

先進入到Flume安裝目錄，執(zhí)行以下第一行命令；
開始新的一個會話窗口，執(zhí)行第二行命令寫入數(shù)據(jù)到指定的文件中
查看上一步驟中指定的文件內(nèi)容

./bin/flume-ng agent -c . -f ./conf/avro.conf -n a1 -Dflume.root.logger=INFO,console
echo 'hello,world' >> ./log.00
bin/flume-ng avro-client --conf conf -H localhost -p 4141 -F ./log.00

執(zhí)行結(jié)果如下

3. 使用netcat數(shù)據(jù)源測試Flume

題目描述

請對Flume的相關配置文件進行設置，從而可以實現(xiàn)如下功能：在一個Linux終端（這里稱為“Flume終端”）中，啟動Flume，在另一個終端（這里稱為“Telnet終端”）中，輸入命令“telnet localhost 44444”，然后，在Telnet終端中輸入任何字符，讓這些字符可以順利地在Flume終端中顯示出來。

編寫Flume配置文件

al.sources = r1
a1.sinks = k1
a1.channels = c1
al.sources.r1.type = netcat
al.sources.r1.channels = c1
a1.sources.r1.bind = localhost
al.sources.r1.port = 44444
a1.sinks.k1.type = logger
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
al.channels.c1.transaction = 100
al.sources.r1.channels = c1
a1.sinks.k1.channel = c1

執(zhí)行以下命令

./bin/flume-ng agent -c . -f ./netcatExample.conf -n a1 -Dflume.root.logger=INFO,console
telnet localhost 44444

會話窗口成功得到數(shù)據(jù)

4. 使用Flume作為Spark Streaming數(shù)據(jù)源

題目描述

Flume是非常流行的日志采集系統(tǒng)，可以作為Spark Streaming的高級數(shù)據(jù)源。請把Flume Source設置為netcat類型，從終端上不斷給Flume Source發(fā)送各種消息，F(xiàn)lume把消息匯集到Sink，這里把Sink類型設置為avro，由Sink把消息推送給Spark Streaming，由自己編寫的Spark Streaming應用程序?qū)ο⑦M行處理。

編寫Flume配置文件

al.sources = r1
a1.sinks = k1
a1.channels =  c1
al.sources.r1.type = netcat
al.sources.r1.bind = localhost
a1.sources.r1.port = 33333
a1.sinks.k1.type = avro
al.sinks.k1.hostname = localhost
a1.sinks.k1.port = 44444
a1.channels.c1.type = memory
al.channels.c1.capacity = 1000000
a1.channels.c1.transactionCapacity = 1000000
al.sources.r1.channels = c1
a1.sinks.k1.channel = c1

主程序代碼

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming._
import org.apache.spark.streaming.Milliseconds
import org.apache.spark.streaming.flume._
import org.apache.spark.util.IntParam
object FlumeEventCount {
    def main(args: Array[String]): Unit = {
        if (args.length < 2) {
            System.err.println( "Usage: FlumeEventCount <host> <port>")
            System.exit(1)
        }
        StreamingExamples.setStreamingLogLevels()
        val Array(host, IntParam(port)) = args
        val batchInterval = Milliseconds(2000)
        val sc = new SparkConf()
          .setAppName("FlumeEventCount")
//          .setMaster("local[2]")
        val ssc = new StreamingContext(sc, batchInterval)
        val stream = FlumeUtils.createStream(ssc, host, port, StorageLevel.MEMORY_ONLY_SER_2)
        stream.count().map(cnt => "Received " + cnt + " flume events." ).print()
        ssc.start()
        ssc.awaitTermination()
    }
}

執(zhí)行結(jié)果1

import org.apache.log4j.{Level, Logger}
import org.apache.spark.internal.Logging
object StreamingExamples extends Logging {
    def setStreamingLogLevels(): Unit = {
        val log4jInitialized = Logger.getRootLogger.getAllAppenders.hasMoreElements
        if (!log4jInitialized) {
            logInfo("Setting log level to [WARN] for streaming example." + " To override add a custom log4j.properties to the classpath.")
            Logger.getRootLogger.setLevel(Level.WARN)
        }
    }
}

執(zhí)行結(jié)果2

以上就是Spark Streaming編程初級實踐詳解的詳細內(nèi)容，更多關于Spark Streaming編程的資料請關注腳本之家其它相關文章！

您可能感興趣的文章:

maven升級版本后報錯:Blocked mirror for repositories
本文主要介紹了maven升級版本后報錯:Blocked mirror for repositories,文中的解決方法非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
2023-09-09
手動部署java項目到k8s中的實現(xiàn)
本文主要介紹了手動部署java項目到k8s中的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2022-08-08
Spring Boot實現(xiàn)Undertow服務器同時支持HTTP2、HTTPS的方法
這篇文章考慮如何讓Spring Boot應用程序同時支持HTTP和HTTPS兩種協(xié)議。小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧
2018-12-12
Java中的static關鍵字全面解析
這篇文章主要介紹了Java中的static關鍵字全面解析的相關資料,非常不錯，具有參考借鑒價值，需要的朋友可以參考下
2016-06-06
使用JDBC工具類實現(xiàn)簡單的登錄管理系統(tǒng)
這篇文章主要為大家詳細介紹了使用JDBC工具類實現(xiàn)簡單的登錄管理系統(tǒng)，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2022-02-02
java多線程并發(fā)executorservice(任務調(diào)度)類
這篇文章主要介紹了線程并發(fā)ScheduledExecutorService類，設置 ScheduledExecutorService ，2秒后，在 1 分鐘內(nèi)每 10 秒鐘蜂鳴一次
2014-01-01
Java語言實現(xiàn)簡單FTP軟件 FTP軟件遠程窗口實現(xiàn)（6）
這篇文章主要為大家詳細介紹了Java語言實現(xiàn)簡單FTP軟件，F(xiàn)TP軟件遠程窗口的實現(xiàn)方法，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2017-03-03
這篇文章主要為大家介紹了Future cancel迷惑性boolean入?yún)⒔馕?，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪
2023-02-02