TOP

Flume+HBase+Kafka集成與開發
2019-01-19 14:11:40 】 瀏覽:461
Tags:

版權聲明:本文為博主原創文章,歡迎大家轉載交流,但必須在博文明顯位置標明轉載地址。否則維權必究! https://blog.csdn.net/py_123456/article/details/83587120


  今天的內容是完成Flume+HBase+Kafka的集成開發。如下圖紅框中所示,節點1的Flume的source有兩個:節點2和節點3的sink輸出。節點1接收后進行預處理然后分別以AsyncHBaseSink(HBaseSink)和Kafka Sink的方式推送給HBase和Kafka進行離線數據處理和實時數據處理。

1.下載Flume源碼并導入Idea開發工具

  1)將apache-flume-1.7.0-src.tar.gz源碼下載到本地解壓

  2)通過idea導入flume源碼

  打開idea開發工具优乐棋牌app下载,選擇File—>Open

  然后找到flume源碼解壓文件,選中flume-ng-hbase-sink,點擊ok加載相應模塊的源碼。

2.官方flume與hbase集成的參數介紹

下的Flume Sink -> AsyncHBaseSink

  其中,加粗的屬性是必須配置的,其它則作為優化參數。payloadColumn屬性是告知HBase有多少個列要寫入列簇columnFamily下。

3.下載日志數據并分析

  到搜狗實驗室用戶查詢日志(該過程在前面HBase環境部署中已經完成,有問題的可以回去查看一下:)

 1)介紹

  搜索引擎查詢日志庫設計為包括約1個月(2008年6月)Sogou搜索引擎部分網頁查詢需求及用戶點擊情況的網頁查詢日志數據集合。為進行中文搜索引擎用戶行為分析的研究者提供基準研究語料

 2)格式說明

  數據格式為:訪問時間\t用戶ID\t[查詢詞]\t該URL在返回結果中的排名\t用戶點擊的順序號\t用戶點擊的URL

  其中,用戶ID是根據用戶使用瀏覽器訪問搜索引擎時的Cookie信息自動賦值,即同一次使用瀏覽器輸入的不同查詢對應同一個用戶ID

  該數據將作為本項目的源數據,存放于節點2和節點3。

4.flume agent-3聚合節點與HBase集成的配置

  用notepad++連接節點1优乐棋牌app下载,對配置文件進行重命名。

  配置fulme-env.sh文件

  配置flume-conf.properties文件

  原模板格式凌亂,直接全部干掉,輸入以下內容:

agent1.sources = r1
agent1.channels = kafkaC hbaseC
agent1.sinks = kafkaSink hbaseSink
agent1.sources.r1.type = avro
agent1.sources.r1.channels = hbaseC
agent1.sources.r1.bind = bigdata-pro01.kfk.com
agent1.sources.r1.port = 5555
agent1.sources.r1.threads = 5
agent1.channels.hbaseC.type = memory
agent1.channels.hbaseC.capacity = 100000
agent1.channels.hbaseC.transactionCapacity = 100000
agent1.channels.hbaseC.keep-alive = 20
agent1.sinks.hbaseSink.type = asynchbase
agent1.sinks.hbaseSink.table = weblogs
agent1.sinks.hbaseSink.columnFamily = info
agent1.sinks.hbaseSink.serializer = org.apache.flume.sink.hbase.KfkAsyncHbaseEventSerializer
agent1.sinks.hbaseSink.channel = hbaseC
agent1.sinks.hbaseSink.serializer.payloadColumn = datatime,userid,searchname,retorder,cliorder,cliurl

5.對日志數據進行格式處理

 1)將文件中的tab更換成逗號

cat weblog.log|tr "\t" "," > weblog2.log

 2)將文件中的空格更換成逗號

cat weblog2.log|tr " " "," > weblog3.log

[kfk@bigdata-pro01 datas]$ rm -f weblog2.log
[kfk@bigdata-pro01 datas]$ rm -f weblog.log
[kfk@bigdata-pro01 datas]$ mv weblog3.log weblog.log
[kfk@bigdata-pro01 datas]$ ls
  wc.input  weblog.log

 3)然后分發到節點2和3

[kfk@bigdata-pro01 datas]$ scp weblog.log bigdata-pro02.kfk.com:/opt/datas/
weblog.log                                                                                                                 100%  145MB  72.5MB/s   00:02   
[kfk@bigdata-pro01 datas]$ scp weblog.log bigdata-pro03.kfk.com:/opt/datas/
weblog.log

6.自定義SinkHBase程序設計與開發

 1)模仿SimpleAsyncHbaseEventSerializer自定義KfkAsyncHbaseEventSerializer實現類优乐棋牌app下载,修改一下代碼即可。

@Override
    public List getActions() {
        List actions = new ArrayList();
        if (payloadColumn != null) {
            byte[] rowKey;
            try {