2. 程式人生 > >大資料求索(9): log4j + flume + kafka + spark streaming實時日誌流處理實戰

大資料求索(9): log4j + flume + kafka + spark streaming實時日誌流處理實戰

阿新 • • 發佈：2018-12-25

大資料求索(9): log4j + flume + kafka + spark streaming實時日誌流處理實戰

一、實時流處理

1.1 實時計算

跟實時系統類似（能在嚴格的時間限制內響應請求的系統），例如在股票交易中，市場資料瞬息萬變，決策通常需要秒級甚至毫秒級。通俗來說，就是一個任務需要在非常短的單位時間內計算出來，這個計算通常是多次的。

1.2 流式計算

通常指源源不斷的資料流過系統，系統能夠不停地連續計算。這裡對時間上可能沒什麼特別限制，資料流入系統到產生結果，可能經過很長時間。比如系統中的日誌資料、電商中的每日使用者訪問瀏覽資料等。

1.3 實時流式計算

將實時計算和流式資料結合起來，就是實時流式計算，也就是大資料中通常說的實時流處理。資料來源源不斷的產生的同時，計算時間上也有了嚴格的限制。比如，目前電商中的商品推薦，往往在你點了某個商品之後，推薦的商品都是變化的，也就是實時的計算出來推薦給你的。再比如你的手機號，在你話費或者流量快用完時，實時的給你推薦流量包套餐等。

二、實時流處理實戰

此例子借鑑慕課網實戰視訊Spark Streaming實時流處理專案實戰，感興趣的可以學習一下。

2.1 源源不斷的資料

此處使用log4j模擬源源不斷產生的日誌資料，啟動一個程序，不停地列印資料即可。簡單配置如下：

log4j.rootLogger=
 
INFO,stdout,flume

log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.target = System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} [%t] [%c] [%p] - %m%n

然後寫一個簡單的日誌列印小程式即可，程式碼如下：

public class LoggerGenerator {
    private static Logger logger = Logger.getLogger(LoggerGenerator.class.getName());

    public static void main(String[] args) throws InterruptedException {
        int index = 0;
        while (true) {
            Thread.sleep(1000);
            logger.info("value : " + index++);
        }
    }
}

2.2 實時採集資料

可以採用Flume實時採集日誌資料，為了和log4j結合，log4j配置檔案需要如下配置

log4j.rootLogger=INFO,stdout,flume

log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.target = System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} [%t] [%c] [%p] - %m%n


log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender
log4j.appender.flume.Hostname = wds
log4j.appender.flume.Port = 41414
log4j.appender.flume.UnsafeMode = true

同時，flume配置如下，

agent1.sources=avro-source
agent1.channels=logger-channel
agent1.sinks=logger-sink

# define source
agent1.sources.avro-source.type=avro
agent1.sources.avro-source.bind=0.0.0.0
agent1.sources.avro-source.port=41414

# define channel
agent1.channels.logger-channel.type=memory

#define sink
agent1.sinks.logger-sink.type = logger

agent1.sources.avro-source.channels=logger-channel
agent1.sinks.logger-sink.channel=logger-channel

這裡暫時採用logger sink， 目的是為了測試資料能否採集到。做專案的過程中，不要想著一步到位，最好做一步測試一步，方便定位錯誤，防止錯誤累積。

啟動flume

 flume-ng agent \
--name agent1 \
--conf $FLUME_HOME/conf \
--conf-file $FLUME_HOME/conf/streaming.conf \
-Dflume.root.logger=INFO,console

注意，這裡可能會報錯，java.lang.ClassNotFoundException: org.apache.flume.clients.log4jappender.Log4jAppender。需要引入jar包來解決此問題

  <dependency>
        <groupId>org.apache.flume.flume-ng-clients</groupId>
        <artifactId>flume-ng-log4jappender</artifactId>
        <version>1.6.0</version>
  </dependency>

這時候啟動日誌列印檔案，可以看到flume已經開始採集資料了。

2018-12-07 21:39:03,204 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{flume.client.log4j.timestamp=1544236744447, flume.client.log4j.logger.name=LoggerGenerator, flume.client.log4j.log.level=20000, flume.client.log4j.message.encoding=UTF8} body: 76 61 6C 75 65 20 3A 20 30                      value : 0 }
2018-12-07 21:39:03,609 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{flume.client.log4j.timestamp=1544236745545, flume.client.log4j.logger.name=LoggerGenerator, flume.client.log4j.log.level=20000, flume.client.log4j.message.encoding=UTF8} body: 76 61 6C 75 65 20 3A 20 31                      value : 1 }
2018-12-07 21:39:04,611 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{flume.client.log4j.timestamp=1544236746548, flume.client.log4j.logger.name=LoggerGenerator, flume.client.log4j.log.level=20000, flume.client.log4j.message.encoding=UTF8} body: 76 61 6C 75 65 20 3A 20 32                      value : 2 }
2018-12-07 21:39:05,614 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{flume.client.log4j.timestamp=1544236747550, flume.client.log4j.logger.name=LoggerGenerator, flume.client.log4j.log.level=20000, flume.client.log4j.message.encoding=UTF8} body: 76 61 6C 75 65 20 3A 20 33                      value : 3 }
2018-12-07 21:39:06,617 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{flume.client.log4j.timestamp=1544236748554, flume.client.log4j.logger.name=LoggerGenerator, flume.client.log4j.log.level=20000, flume.client.log4j.message.encoding=UTF8} body: 76 61 6C 75 65 20 3A 20 34                      value : 4 }
2018-12-07 21:39:07,620 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{flume.client.log4j.timestamp=1544236749556, flume.client.log4j.logger.name=LoggerGenerator, flume.client.log4j.log.level=20000, flume.client.log4j.message.encoding=UTF8} body: 76 61 6C 75 65 20 3A 20 35                      value : 5 }
2018-12-07 21:39:08,626 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{flume.client.log4j.timestamp=1544236750560, flume.client.log4j.logger.name=LoggerGenerator, flume.client.log4j.log.level=20000, flume.client.log4j.message.encoding=UTF8} body: 76 61 6C 75 65 20 3A 20 36                      value : 6 }
2018-12-07 21:39:09,632 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{flume.client.log4j.timestamp=1544236751568, flume.client.log4j.logger.name=LoggerGenerator, flume.client.log4j.log.level=20000, flume.client.log4j.message.encoding=UTF8} body: 76 61 6C 75 65 20 3A 20 37                      value : 7 }
2018-12-07 21:39:10,636 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{flume.client.log4j.timestamp=1544236752572, flume.client.log4j.logger.name=LoggerGenerator, flume.client.log4j.log.level=20000, flume.client.log4j.message.encoding=UTF8} body: 76 61 6C 75 65 20 3A 20 38                      value : 8 }
2018-12-07 21:39:11,638 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{flume.client.log4j.timestamp=1544236753575, flume.client.log4j.logger.name=LoggerGenerator, flume.client.log4j.log.level=20000, flume.client.log4j.message.encoding=UTF8} body: 76 61 6C 75 65 20 3A 20 39                      value : 9 }
2018-12-07 21:39:12,640 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{flume.client.log4j.timestamp=1544236754577, flume.client.log4j.logger.name=LoggerGenerator, flume.client.log4j.log.level=20000, flume.client.log4j.message.encoding=UTF8} body: 76 61 6C 75 65 20 3A 20 31 30                   value : 10 }

2.3 訊息佇列緩衝資料

這裡flume採集到的資料將存到kafka中，作為一個緩衝，也即生產者。此時，需要將flume和kafka打通，flume配置修改如下：

 agent1.sources=avro-source
 agent1.channels=logger-channel
 agent1.sinks=kafka-sink
 
 # define source
 agent1.sources.avro-source.type=avro
 agent1.sources.avro-source.bind=0.0.0.0
 agent1.sources.avro-source.port=41414

 # define channel
 agent1.channels.logger-channel.type=memory

 #define sink
 agent1.sinks.kafka-sink.type = org.apache.flume.sink.kafka.KafkaSink
 agent1.sinks.kafka-sink.brokerList = wds:9092
 agent1.sinks.kafka-sink.topic = streamingtopic
 agent1.sinks.kafka-sink.batchSize = 20
 agent1.sinks.kafka-sink.requiredAcks = 1
 
 agent1.sources.avro-source.channels=logger-channel
 agent1.sinks.kafka-sink.channel=logger-channel

kafka依賴zookeeper，需要先啟動zookeeper

$ZK_HOME/bin/zkServer.sh start

啟動kafka，需要指定配置檔案

 bin/kafka-server-start.sh config/server.properties

建立topic

kafka-topics.sh --create --zookeeper wds:2181 --replication-factor 1 --partitions 1 --topic streamingtopic

啟動消費程序，測試消費資訊測試

kafka-console-consumer.sh --zookeeper wds:2181  --topic streamingtopic

可以看到每隔20個(由flume配置檔案裡的agent1.sinks.kafka-sink.batchSize = 20決定的)有消費資訊輸出，證明連線成功

[[email protected] ~]$  kafka-console-consumer.sh --zookeeper wds:2181  --topic streamingtopic
value : 0
value : 1
value : 2
value : 3
value : 4
value : 5
value : 6
value : 7
value : 8
value : 9
value : 10
value : 11
value : 12
value : 13
value : 14
value : 15
value : 16
value : 17
value : 18
value : 19

2.4 實時處理資料

使用Spark Streaming從Kafka消費訊息，這裡採用Receiver模式，可以參照下面簡單的程式碼

object KafkaReceiverWordCount {
  def main(args: Array[String]): Unit = {
    if (args.length != 4) {
      System.err.println("Usage: KafkaReceiverWordCount <ZkQuorum> <group> <topics> <numThreads>")
      System.exit(1)
    }
    val Array(zkQuorum, group, topics, numThreads) = args

    val sparkConf = new SparkConf().setAppName("KafkaReceiverWordCount").setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
    // Kafka對接Spark Streaming
    val messages = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap)
    // _1引數沒有用
    messages.map(_._2).count()

    ssc.start()
    ssc.awaitTermination()

  }
}

注意，日誌和Streaming日誌接收都是在本地的，那麼生產環境如何做呢？

1） 打包jar，指向LoggerGenerator類
2） Flume和Kafka一樣
3） Spark Streaming也需要打成jar包，然後提交到叢集群執行，Spark-submit方式提交執行，模式為local/yarn/standalone/mesos

jar包使用maven的mvn assembly:assembly -Dmaven.test.skip=true方式打包，把kafka相關jar包也打進去，不需要的使用provided
打包的時候需要將LoggerGenerator從test中移出來放到java的某個包下，方便執行

打包成功後，在服務上執行

java -cp spark-test-1.0-jar-with-dependencies.jar com.wds.streaming.LoggerGenerator

其中-cp命令是將xxx.jar加入到classpath，這樣java class loader就會在這裡面查詢匹配的類，這樣打包的時候不用指定main class，非常方便。然後提交jar包到spark-submit

spark-submit \
--class com.wds.streaming.KafkaStreamingApp \
--master local[2] \
--name KafkaStreamingApp \
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.2 \
/home/hadoop/hadoop/lib/spark-test-1.0-jar-with-dependencies.jar wds:2181 test streamingtopic 2

日誌輸出如下

[[email protected] lib]$ java -cp spark-test-1.0-jar-with-dependencies.jar com.wds.streaming.LoggerGenerator
log4j:ERROR Could not find value for key log4j.appender.flume.layout
2018-12-07 22:32:27,447 [main] [org.apache.flume.api.NettyAvroRpcClient] [WARN] - Using default maxIOWorkers
2018-12-07 22:32:28,681 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 0
2018-12-07 22:32:29,751 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 1
2018-12-07 22:32:30,753 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 2
2018-12-07 22:32:31,755 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 3
2018-12-07 22:32:32,758 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 4
2018-12-07 22:32:33,760 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 5
2018-12-07 22:32:34,762 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 6
2018-12-07 22:32:35,764 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 7
2018-12-07 22:32:36,765 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 8
2018-12-07 22:32:37,767 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 9
2018-12-07 22:32:38,770 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 10
2018-12-07 22:32:39,772 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 11
2018-12-07 22:32:40,775 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 12
2018-12-07 22:32:41,777 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 13
2018-12-07 22:32:42,779 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 14
2018-12-07 22:32:43,782 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 15
2018-12-07 22:32:44,784 [main] [com.wds.streaming.LoggerGenerator] [INFO] - value : 16

Spark Streaming輸出如下:

18/12/07 22:32:50 INFO executor.Executor: Running task 0.0 in stage 12.0 (TID 10)
18/12/07 22:32:50 INFO storage.ShuffleBlockFetcherIterator: Getting 0 non-empty blocks out of 2 blocks
18/12/07 22:32:50 INFO storage.ShuffleBlockFetcherIterator: Started 0 remote fetches in 1 ms
18/12/07 22:32:50 INFO executor.Executor: Finished task 0.0 in stage 12.0 (TID 10). 1705 bytes result sent to driver
18/12/07 22:32:50 INFO scheduler.TaskSetManager: Finished task 0.0 in stage 12.0 (TID 10) in 6 ms on localhost (executor driver) (1/1)
18/12/07 22:32:50 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 12.0, whose tasks have all completed, from pool 
18/12/07 22:32:50 INFO scheduler.DAGScheduler: ResultStage 12 (print at KafkaStreamingApp.scala:23) finished in 0.006 s
18/12/07 22:32:50 INFO scheduler.DAGScheduler: Job 6 finished: print at KafkaStreamingApp.scala:23, took 0.014111 s
-------------------------------------------
Time: 1544239970000 ms
-------------------------------------------
20

至此，一切打通，測試成功。

三、參考

  1. Spark Streaming實時流處理專案實戰
  2. flume、kafka、spark官方文件

相關推薦

大資料求索(9): log4j + flume + kafka + spark streaming實時日誌流處理實戰

大資料求索(9): log4j + flume + kafka + spark streaming實時日誌流處理實戰 一、實時流處理 1.1 實時計算 跟實時系統類似（能在嚴格的時間限制內響應請求的系統），例如在股票交易中，市場資料瞬息萬變，決策通常需要秒級甚至毫秒級。通俗來

Flume+Kafka+Spark Streaming實現大資料實時流式資料採集

大資料實時流式資料處理是大資料應用中最為常見的場景，與我們的生活也息息相關，以手機流量實時統計來說，它總是能夠實時的統計出使用者的使用的流量，在第一時間通知使用者流量的使用情況，並且最為人性化的為使用者提供各種優惠的方案，如果採用離線處理，那麼等到使用者流量超標

基於Flume+Kafka+Spark Streaming打造實時流處理項目實戰課程

大數據本課程從實時數據產生和流向的各個環節出發，通過集成主流的分布式日誌收集框架Flume、分布式消息隊列Kafka、分布式列式數據庫HBase、及當前最火爆的Spark Streaming打造實時流處理項目實戰，讓你掌握實時處理的整套處理流程，達到大數據中級研發工程師的水平！下載地址:百度網盤下載

帶你看懂大資料採集引擎之Flume&採集目錄中的日誌

歡迎關注大資料和人工智慧技術文章釋出的微信公眾號：清研學堂，在這裡你可以學到夜白（作者筆名）精心整理的筆記，讓我們每天進步一點點，讓優秀成為一種習慣！ 帶你看懂大資料採集引擎之Flume&採集目錄中的日誌 一、Flume的介紹： Flume由Clo

基於 Flume+Kafka+Spark Streaming 實現實時監控輸出日誌的報警系統

運用場景：我們機器上每天或者定期都要跑很多工，很多時候任務出現錯誤不能及時發現，導致發現的時候任務已經掛了很久了。  解決方法：基於 Flume+Kafka+Spark Streaming 的框架對這些任務的輸出日誌進行實時監控，當檢測到日誌出現Error的資訊就傳送郵件給

flume+kafka+spark streaming(持續更新)

kafka Kafka是一種高吞吐量的分散式釋出訂閱訊息系統，它可以處理消費者規模的網站中的所有動作流資料。 kafka的設計初衷是希望作為一個統一的資訊收集平臺,能夠實時的收集反饋資訊,並需要能夠支撐較大的資料量,且具備良好的容錯能力. Apache

使用Flume+Kafka+SparkStreaming進行實時日誌分析

每個公司想要進行資料分析或資料探勘，收集日誌、ETL都是第一步的，今天就講一下如何實時地（準實時，每分鐘分析一次）收集日誌，處理日誌，把處理後的記錄存入Hive中，並附上完整實戰程式碼 1. 整體架構 思考一下，正常情況下我們會如何收集並分析日誌呢？

大資料學習之路97-kafka直連方式（spark streaming 整合kafka 0.10版本）

我們之前SparkStreaming整合Kafka的時候用的是傻瓜式的方式-----createStream,但是這種方式的效率很低。而且在kafka 0.10版本之後就不再提供了。 接下來我們使用Kafka直連的方式，這種方式其實是呼叫Kafka底層的消費資料的API,我們知道，越底層的東

大資料求索(8):Spark Streaming簡易入門一

大資料求索(8):Spark Streaming簡易入門一 一、Spark Streaming簡單介紹 Spark Streaming是基於Spark Core上的一個應用程式，可伸縮，高吞吐，容錯（這裡主要是藉助Spark Core的容錯方式）處理線上資料流，資料可以有不同的來

# 大資料求索(7)： Kafka的重要原理和概念二與實戰

大資料求索(7)： Kafka的重要原理和概念二 大資料最好的學習資料是官方文件。 Kafka官方文件地址：http://kafka.apache.org/ 四、Kafka高效性相關設計 4.1 訊息的持久化 Kafka高度依賴檔案系統來儲存和快取訊息

大資料求索(7)： Kafka的重要原理和概念一

大資料求索(7)： Kafka的重要原理和概念一 大資料最好的學習資料是官方文件。 Kafka官方文件地址：http://kafka.apache.org/ 一、Kakfa簡介 Apache kafka 是一個分散式的基於push-subscribe的訊息

大資料求索(5)：Windows下使用IDEA開發Kafka程式和伺服器通訊失敗問題

問題描述 在windows下使用IDEA開發完生產者程式以後，在伺服器端開啟消費者程序，發現傳送失敗，無法進行通訊，報錯如下： kafka .FailedToSendMessageException: Failed to send messages after 3 tries

大資料求索（4）：配置Spark History Server

大資料求索（4）：配置Spark History Server 修改配置檔案 修改conf/spark-default.conf cp spark-default.conf.template spark-default.conf vim spark-defau

大資料學習之路94-kafka叢集安裝

解壓 Kafka 安裝包 修改配置檔案 config/server.properties vi server.properties broker.id=0 //為依次增長的：0、1、2、3、4，叢集中唯一id log.dirs=/kafkaData/logs // Kafka

大資料學習之路93-kafka簡介

kafka是實時計算中用來做資料收集的，它是一個訊息佇列。它使用scala開發的。 那麼我們就會想我們這裡能不能用hdfs做資料儲存呢？它是分散式的，高可用的。 但是它還缺少一些重要的功能：比如說我們往hdfs中寫資料，之後我們需要實時的讀取。當我們讀到某一行的時候斷掉了，假如說這個讀取

大資料入門（20）kafka安裝配置

kafka基本概念 1、kafka是一個分散式的訊息快取系統 2、kafka叢集中的伺服器都叫做broker 3、kafka有兩類客戶端，一類叫producer（訊息生產者），一類叫做consumer（訊息消費者），客戶端和broker伺服器之間採用tcp協議連線 4、kafka中不同業務系統的

大資料求索(3):實戰MapReduce

MapReduce 概述 主要用於離線、海量資料運算 WordCount編寫 下面這張經典圖很好地說明了如何編寫一個WordCount，也清楚說明了MapReduce的流程 對於輸入的一個文字（可以存放在HDFS上，可以非常非常大），先對檔案進行拆分，假設這裡一行一

大資料求索(2):yarn的配置與使用

yarn的架構 1） Resourcemanager: RM 整個叢集同一時間提供服務的RM只有一個，負責叢集資源的統一管理和排程 處理來自客戶端的請求，比如提交一個job，殺死一個job等 監控NM，一旦某個NM掛了，那麼該NM上執行的任務需要告訴AM來如何

大資料求索(1)：HDFS安裝指南

大資料求索(1)：HDFS安裝指南 背景 HDFS是分散式檔案系統，是Hadoop的基礎，本章主要介紹如何安裝HDFS。 環境引數 centos 6.5 hadoop2.6.0-cdh5.7.0 jdk8 偽分散式安裝 一、JDK安裝

大資料求索(4)：Hive安裝和使用

Hive安裝和使用 Hive環境搭建 Hive下載 wget https://archive.cloudera.com/cdh5/cdh/5/hive-1.1.0-cdh5.7.0.tar.gz 解壓 tar -zxvf hive-1.1.00cdh5.7.0