您好,登錄后才能下訂單哦!
這篇“Spark Streaming編程初級源碼分析”文章的知識點大部分人都不太理解,所以小編給大家總結了以下內容,內容詳細,步驟清晰,具有一定的借鑒價值,希望大家閱讀完這篇文章能有所收獲,下面我們一起來看看這篇“Spark Streaming編程初級源碼分析”文章吧。
Linux:CentOS7.5
Spark: spark-3.0.0-bin-hadoop3.2
Flume:Flume-
1.9.0
IDE:IntelliJ IDEA2020.2.3
Flume是Cloudera提供的一個分布式、可靠、可用的系統,它能夠將不同數據源的海量日志數據進行高效收集、聚合、移動,最后存儲到一個中心化數據存儲系統中。Flume 的核心是把數據從數據源收集過來,再送到目的地。
或者也可以直接到本教程官網的“下載專區”中的“軟件”目錄中下載apache-flume-1.7.0-bin.tar.gz。
下載后,把Flume1.7.0安裝到Linux系統的“/usr/local/flume”目錄下,具體安裝和使用方法可以參考教程官網的“實驗指南”欄目中的“日志采集工具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
Avro可以發送一個給定的文件給Flume,Avro 源使用AVRO RPC機制。請對Flume的相關配置文件進行設置,從而可以實現如下功能:在一個終端中新建一個文件helloworld.txt(里面包含一行文本“Hello World”),在另外一個終端中啟動Flume以后,可以把helloworld.txt中的文本內容顯示出來。
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
先進入到Flume安裝目錄,執行以下第一行命令;
開始新的一個會話窗口,執行第二行命令寫入數據到指定的文件中
查看上一步驟中指定的文件內容
./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
請對Flume的相關配置文件進行設置,從而可以實現如下功能:在一個Linux終端(這里稱為“Flume終端”)中,啟動Flume,在另一個終端(這里稱為“Telnet終端”)中,輸入命令“telnet localhost 44444”,然后,在Telnet終端中輸入任何字符,讓這些字符可以順利地在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
執行以下命令
./bin/flume-ng agent -c . -f ./netcatExample.conf -n a1 -Dflume.root.logger=INFO,console telnet localhost 44444
會話窗口成功得到數據
Flume是非常流行的日志采集系統,可以作為Spark Streaming的高級數據源。請把Flume Source設置為netcat類型,從終端上不斷給Flume Source發送各種消息,Flume把消息匯集到Sink,這里把Sink類型設置為avro,由Sink把消息推送給Spark Streaming,由自己編寫的Spark Streaming應用程序對消息進行處理。
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() } }
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) } } }
以上就是關于“Spark Streaming編程初級源碼分析”這篇文章的內容,相信大家都有了一定的了解,希望小編分享的內容對大家有幫助,若想了解更多相關的知識內容,請關注億速云行業資訊頻道。
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。