2. 程式人生 > 實用技巧 >Flume+Kafka+Sparkstreaming日誌分析

Flume+Kafka+Sparkstreaming日誌分析

阿新 • • 發佈：2020-10-22

2019獨角獸企業重金招聘Python工程師標準>>> hot3.png

最近要做一個日誌實時分析的應用，採用了flume+kafka+sparkstreaming框架，先搞了一個測試Demo，本文沒有分析其架構原理。

　　簡介：flume是一個分散式，高可靠，可用的海量日誌聚合系統，kafka是一高吞吐量的分散式釋出訂閱系統，sparkstreaming是建立在spark上的實時計算框架，這這個Demo中，以上內容均為單機版偽分佈，flume的source為exec，agent的名稱為producer，sink為kafka。

　　執行所需要的環境直接到官網上下載即可：

　　我的環境是：flume1.6+kafka_2.10+spark1.2.0

　　flume的配置：

　　在conf下編輯配置檔案roomy.conf如下：

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

#agent section

producer.sources = s

producer.channels = c

producer.sinks = r

#source section

producer.sources.s.type=exec

producer.sources.s.command=tail-F -n+1/Users/roomy/Desktop/Coding/scala/real_time_project/debug

.log#監聽日誌所在

producer.sources.s.channels = c

# Each sink's type must be defined

producer.sinks.r.type= org.apache.flume.plugins.KafkaSink

producer.sinks.r.metadata.broker.list=192.168.1.102:9092#這裡換成自己Kafka的地址

producer.sinks.r.partition.key=0

producer.sinks.r.partitioner.class=org.apache.flume.plugins.SinglePartition

producer.sinks.r.serializer.class=kafka.serializer.StringEncoder

producer.sinks.r.request.required.acks=0

producer.sinks.r.max.message.size=1000000

producer.sinks.r.producer.type=sync

producer.sinks.r.custom.encoding=UTF-8

　　在flume資料夾下執行

1

bin/flume-ngagent --conf conf --conf-fileconf/roomy.conf --name producer -Dflume.root.logger=INFO,console

　　flume的部分完成。

　　在kafka目錄下執行：

1

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

　　啟動zookeeper

　　執行：

1

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

　　啟動kafka，這裡無需做什麼額外配置。

　　最後編寫spark streaming測試Demo程式

　　直接新建SBT專案，build.sbt如下：

1

2

3

4

5

6

7

8

9

10

11

name :="sk"

version :="1.0"

scalaVersion :="2.10.4"

libraryDependencies +="org.apache.spark"%"spark-streaming_2.10"%"1.6.1"

libraryDependencies +="org.apache.spark"%"spark-streaming-kafka_2.10"%"1.6.1"

libraryDependencies +="log4j"%"log4j"%"1.2.17"

　　需要注意的是，由於GFW，下載慢的要死，接下來就是測試程式

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

importkafka.serializer.StringDecoder

importorg.apache.spark.SparkConf

importorg.apache.spark.streaming.dstream.DStream

importorg.apache.spark.streaming.{Seconds, StreamingContext}

importorg.apache.spark.streaming.kafka.KafkaUtils

/**

* Created by roomy on 16/3/23.

*/

objectKafkaStreaming {

defmain(agrs:Array[String]):Unit={

valsparkConf=newSparkConf().setMaster("local[2]").setAppName("Streamingtest")

valssc=newStreamingContext(sparkConf, Seconds(20))

valtopic="test"

valtopicSet=topic.split(" ").toSet

//create direct kafka stream with brokers and topics

valkafkaParams=Map[String, String]("metadata.broker.list"->"localhost:9092")

valmessages=KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](

ssc, kafkaParams, topicSet

)

vallines=messages.map(_._2)

lines.print()

valwords:DStream[String]=lines.flatMap(_.split("\n"))

words.count().print()

//啟動

ssc.start()

ssc.awaitTermination()

}

}

　　可以通過StreamContext的建構函式設定資料採集分析的間隔。

　　程式會監聽/Users/roomy/Desktop/Coding/scala/real_time_project/debug.log中的變動，並以20秒一次的頻率總計增加行數輸出在控制檯。

　　日誌沒有變動的時候如下：

　　執行測試程式產生日誌：

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

importorg.apache.log4j.Logger;

/**

* Created by roomy on 16/3/23.

* to generate some log to test

*/

publicclassLogGeneratorimplementsRunnable{

privatestaticLogger logger = Logger.getLogger(LogGenerator.class);

privateintno;

publicLogGenerator(intno){

this.no=no;

}

publicstaticvoidmain(String [] agrs)throwsInterruptedException {

for(inti=0;i<5;i++){

newThread(newLogGenerator(i)).start();

}

}

@Override

publicvoidrun() {

while(true){

logger.debug("this is a test information produced by roomy no:"+Thread.currentThread().getName());

try{

Thread.sleep((int)Math.random()*100);

}

catch(Exception e){

e.printStackTrace();

}

}

}

}

　　控制檯輸出如下：

　　streaming的輸出操作會把每個批次的前十個元素輸出如下：

　　在這20秒內總共產生的日誌行數為：

　　參考文件：

　　https://flume.apache.org/FlumeUserGuide.html

　　http://kafka.apache.org/documentation.html　

　　Spark快速大資料分析

轉載於:https://my.oschina.net/hblt147/blog/1840271

相關推薦

Flume+Kafka+Sparkstreaming日誌分析

2019獨角獸企業重金招聘Python工程師標準>>> 最近要做一個日誌實時分析的應用，採用了flume+kafka+sparkstreaming框架，先搞了一個測試Demo，本文沒有分析其架構原理。

網站流量日誌分析（資料 採集之 Flume 採集）

目錄網站流量日誌分析的意義如何進行網站分析流量分析內容導航分析轉化分析（漏斗模型分析）網站流量日誌分析的資料處理流程資料採集資料預處理資料入庫資料分析資料視覺化埋點資料採集資料採集方式及其優缺點埋點js

企業級實戰模組二：ELK+Filebeat+Kafka+ZooKeeper構建大資料日誌分析平臺案例（上）

企業級實戰模組二：ELK+Filebeat+Kafka+ZooKeeper構建大資料日誌分析平臺案例（上）

企業級實戰模組三：ELK+Filebeat+Kafka+ZooKeeper構建大資料日誌分析平臺案例（下）

企業級實戰模組三：ELK+Filebeat+Kafka+ZooKeeper構建大資料日誌分析平臺案例（下）

fliebeat+kafka的ELK日誌分析平臺

目錄一.環境講解二.部署配置配置kafka叢集配置日誌輸出端配置logstash配置elasticsearch配置kibana四.kibana圖形操作建立索引設定圖形展示

專案-日誌分析平臺

日誌分析平臺(練手專案) 練習hdfs mr hive hbase 各種公司都需要，例如電商、旅遊(攜程)、保險種種。

關於Mysql通用查詢日誌和慢查詢日誌分析

MySQL中的日誌包括：錯誤日誌、二進位制日誌、通用查詢日誌、慢查詢日誌等等。這裡主要介紹下比較常用的兩個功能：通用查詢日誌和慢查詢日誌。

nginx伺服器中access_log日誌分析與配置詳解

前言 nginx的log日誌分為：access log 和 error log 其中access log 記錄了哪些使用者，哪些頁面以及使用者瀏覽器、ip和其他的訪問資訊

GoAccess日誌分析工具

1.1 GoAccess簡介 GoAccess是一個非常良心的開源軟體，它的良心之處體現在如下方面：1）安裝簡單；2）操作容易；3）介面酷炫；GoAccess 官網https://goaccess.io

AWStats日誌分析系統

1.什麼是AWStats AWStats是一款功能強大且功能強大的免費工具，可以圖形方式生成高階Web，流媒體，ftp或郵件伺服器統計資訊。此日誌分析器用作CGI或命令列， 並在幾個圖形網頁中顯示您的日誌包含的所有可能資訊。它使

【C++】一個實用的日誌分析工具

【C++】一個實用的日誌分析工具 程式說明 通過C語言巨集定義獲取時間、檔案、函式與行數；通過可變引數列表實現日誌內容的傳遞；C語言巨集定義包括：__FILE__、__FUNCTION__、__LINE__、__DATA__、__TIME__

網站流量日誌分析-01

網站流量日誌分析-01 網站流量日誌分析的意義 通過分析使用者的行為資料 讓更多的使用者沉澱下來變成會員 賺取更多的錢。

搭建視覺化日誌分析平臺ELK

https://blog.csdn.net/wangkai_123456/article/details/84796903 https://blog.csdn.net/qq942477618/article/details/52951011?utm_medium=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai

網站流量日誌分析（模組開發——資料倉庫設計）

目錄數倉設計維度建模本專案中資料倉庫的設計事實表設計原始資料表: ods_weblog_origin =>對應mr清洗完之後的資料訪問日誌明細寬表：dw_weblog_detail維度表設計多維度資料分析維度建模的三種模式本專案模式設計資料

網站流量日誌分析（模組開發——資料預處理）

目錄資料預處理預處理的程式設計思路問題MapReduce程式設計技巧點選流模型的概述會話（session）程式碼pom.xmllog4j.propertiespreprocess 模組WebLogBeanWebLogMainWebLogMapperpageviews 模組ClickStreamPageViewP

大資料場景下資料異構之 Mysql實時寫入HBase（藉助canal kafka SparkStreaming）

背景：公司線下ETC機房有個Mycat叢集，供訂單系統使用，現需要進行資料異構將Mysql資料(近)實時寫入另一套資料庫用作讀請求和資料歸檔用

Tomcat日誌分析

1 #!/bin/bash 2 #Shellstatistics.sh 3 #Author 　　 lipc 4 #Date　20200818 5 #Ps　　 tomcat接入日誌分析,需要tomcat日誌開啟訪問時間記錄

ELK日誌分析服務搭建

本實驗所用軟體包下載地址：連結：https://pan.baidu.com/s/1h4q8tL5fJ8TqcCzhHwAVpg 提取碼：8oaf

kafka學習筆記（四）kafka的日誌模組

概述 日誌段及其相關程式碼是 Kafka 伺服器原始碼中最為重要的元件程式碼之一。你可能會非常關心，在 Kafka 中，訊息是如何被儲存和組織在一起的。畢竟，不管是學習任何訊息引擎，弄明白訊息建模方式都是首要的問題。

openresty+lua+kafka實現日誌收集

本文轉自：https://www.cnblogs.com/gxyandwmm/p/11298912.html ********************* 部署過程 **************************