2. 程式人生 > >flume讀取日誌資料寫入kafka 然後kafka+storm整合

flume讀取日誌資料寫入kafka 然後kafka+storm整合

阿新 • • 發佈：2019-01-10

一、flume配置

flume要求1.6以上版本

flume-conf.properties檔案配置內容，sinks的輸出作為kafka的product

  1. a1.sources = r1
  2. a1.sinks = k1
  3. a1.channels = c1
  4. # Describe/configure the source  
  5. a1.sources.r1.type = exec
  6. a1.sources.r1.command = tail -F /home/airib/work/log.log  
  7. # Describe the sink  
  8. #a1.sinks.k1.type
     = logger
  9. a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink  
  10. a1.sinks.k1.topic = test
  11. a1.sinks.k1.brokerList = localhost:9092  
  12. a1.sinks.k1.requiredAcks = 1
  13. a1.sinks.k1.batchSize = 20
  14. # Use a channel which buffers events in memory  
  15. a1.channels.c1.type = memory
  16. a1.channels.c1.capacity = 
    1000
  17. a1.channels.c1.transactionCapacity = 100
  18. # Bind the source and sink to the channel  
  19. a1.sources.r1.channels = c1
  20. a1.sinks.k1.channel = c1

flume啟動

 bin/flume-ng agent --conf conf --conf-file conf/flume-conf.properties --name a1 -Dflume.root.logger=INFO,console

二  kafka的消費者java原始碼
  1. package com.hgp.kafka.kafka;  
  2. import java.util.HashMap;  
  3. import java.util.List;  
  4. import java.util.Map;  
  5. import java.util.Properties;  
  6. import kafka.consumer.ConsumerConfig;  
  7. import kafka.consumer.ConsumerIterator;  
  8. import kafka.consumer.KafkaStream;  
  9. import kafka.javaapi.consumer.ConsumerConnector;  
  10. import kafka.serializer.StringDecoder;  
  11. import kafka.utils.VerifiableProperties;  
  12. public class KafkaConsumer {  
  13.     private final ConsumerConnector consumer;  
  14.     private KafkaConsumer() {  
  15.         Properties props = new Properties();  
  16.         //zookeeper 配置  
  17.         props.put("zookeeper.connect", "localhost:2181");  
  18.         //group 代表一個消費組  
  19.         props.put("group.id", "jd-group");  
  20.         //zk連線超時  
  21.         props.put("zookeeper.session.timeout.ms", "4000");  
  22.         props.put("zookeeper.sync.time.ms", "200");  
  23.         props.put("auto.commit.interval.ms", "1000");  
  24.         props.put("auto.offset.reset", "smallest");  
  25.         //序列化類  
  26.         props.put("serializer.class", "kafka.serializer.StringEncoder");  
  27.         ConsumerConfig config = new ConsumerConfig(props);  
  28.         consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);  
  29.     }  
  30.     void consume() {  
  31.         Map<String, Integer>topicCountMap = new HashMap<String, Integer>();  
  32.         topicCountMap.put("test", new Integer(1));  
  33.         StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());  
  34.         StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());  
  35.         Map<String, List<KafkaStream<String, String>>>consumerMap =   
  36.                 consumer.createMessageStreams(topicCountMap,keyDecoder,valueDecoder);  
  37.         KafkaStream<String, String>stream = consumerMap.get("test").get(0);  
  38.         ConsumerIterator<String, String>it = stream.iterator();  
  39.         while (it.hasNext())  
  40.             System.out.println(it.next().message());  
  41.     }  
  42.     public static void main(String[] args) {  
  43.         new KafkaConsumer().consume();  
  44.     }  
  45. }  

kafka啟動命令

啟動Zookeeper server： 

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

啟動Kafka server:

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

執行producer： 

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

執行consumer： 

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning 

二、示例

  1. package com.hgp.kafka.kafka;  
  2. import java.util.Arrays;  
  3. import java.util.HashMap;  
  4. import java.util.Iterator;  
  5. import java.util.Map;  
  6. import java.util.Map.Entry;  
  7. import java.util.concurrent.atomic.AtomicInteger;  
  8. import org.apache.commons.logging.Log;  
  9. import org.apache.commons.logging.LogFactory;  
  10. import storm.kafka.BrokerHosts;  
  11. import storm.kafka.KafkaSpout;  
  12. import storm.kafka.SpoutConfig;  
  13. import storm.kafka.StringScheme;  
  14. import storm.kafka.ZkHosts;  
  15. import backtype.storm.Config;  
  16. import backtype.storm.LocalCluster;  
  17. import backtype.storm.StormSubmitter;  
  18. import backtype.storm.generated.AlreadyAliveException;  
  19. import backtype.storm.generated.InvalidTopologyException;  
  20. import backtype.storm.spout.SchemeAsMultiScheme;  
  21. import backtype.storm.task.OutputCollector;  
  22. import backtype.storm.task.TopologyContext;  
  23. import backtype.storm.topology.OutputFieldsDeclarer;  
  24. import backtype.storm.topology.TopologyBuilder;  
  25. import backtype.storm.topology.base.BaseRichBolt;  
  26. import backtype.storm.tuple.Fields;  
  27. import backtype.storm.tuple.Tuple;  

  28. 相關推薦

    flume讀取日誌資料寫入kafka 然後kafka+storm整合

    一、flume配置 flume要求1.6以上版本 flume-conf.properties檔案配置內容，sinks的輸出作為kafka的product a1.sources = r1 a1.sinks = k1 a1.cha

    flume讀取日誌資料寫入kafka

    一、flume配置 flume要求1.6以上版本 flume-conf.properties檔案配置內容，sinks的輸出作為kafka的product a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/

    Flume讀取日誌資料並寫入到Kafka，ConsoleConsumer進行實時消費

    最近大資料學習使用了Flume、Kafka等，今天就實現一下Flume實時讀取日誌資料並寫入到Kafka中，同時，讓Kafka的ConsoleConsumer對日誌資料進行消費。 1、Flume F

    電商大資料分析平臺（三）nginx配置及flume讀取日誌資訊

    一、nginx配置 在本專案中nginx的作用只是接收客戶端傳送的事件，並將相應的session寫入日誌檔案中，所以配置較為簡單，只需要配置寫入的日誌檔案和寫入的格式 1.地址配置 server { listen

    程式碼 | Spark讀取mongoDB資料寫入Hive普通表和分割槽表

    版本： spark 2.2.0  hive 1.1.0  scala 2.11.8  hadoop-2.6.0-cdh5.7.0  jdk 1.8  MongoDB 3.6.4 一 原始資料及Hive表  MongoDB資

    Flume+hbase 日誌資料採集與儲存

    瞭解過flume的人，差不多都看過這張或則類似的圖片，本文即實現上圖部分內容。（由於條件有限，目前是單機上實現） flume-agent配置檔案 #flume agent conf source_agent.sources = server source_agent.si

    Flume實現日誌資料夾資料載入到HDFS

    Flume是一種分散式，可靠和可用的服務，用於高效收集，聚合和移動大量日誌資料。 它具有基於資料流的簡單和可伸縮的架構。 它具有可靠性機制和故障切換和恢復機制的魯棒性和容錯能力。 vi corp_base_info.conf a1.sources = r1

    Apache Kafka -7 與Storm整合

    Apache Kafka教程 之 與Storm整合 Apache Kafka - 與Storm整合 關於Storm Storm最初是由Nathan Marz和BackType建立的。在短時間內，Apache Storm成為分散式實時

    filebeat讀取nginx日誌並寫入kafka

    com res 讀取 code inpu file max eat compress filebeat寫入kafka的配置： filebeat.inputs: - type: log paths: - /tmp/access.log tags: ["ngin

    通過flume把日誌檔案內容寫入kafka主題

    首先自行安裝flume和 kafka當然還要jdk，我flume版本是1.6的kafka版本2.11，jdk1.8。 首先在路徑flume下的conf裡面建立一個logtokafka.conf檔案進行配置配置內容如下。 agent.sources=r1 agent.sinks=k1 agen

    [大資料] 搜尋日誌資料採集系統 flume+hbase+kafka架構 （資料搜狗實驗室）

    1 採集規劃 說明： D1 日誌所在伺服器1 —bigdata02.com D2 日誌所在伺服器2 —bigdata03.com 日誌收集 日誌收集 日誌整合 儲存到kafka 儲存到HBase 2版本 kafka kafka_2.11-0.10

    Flume, Kafka和NiFi，大資料實時日誌資料收集、資料獲取技術哪家強？

    作者Tony Siciliani  本文為36大資料獨譯，譯者：隨風 我們在建設一個大資料管道時，需要在Hadoop生態系統前仔細考慮，如何獲取大體量、多樣化以及高速性的資料。在決定採用何種工具以滿足我們的需求時，最初對於擴充套件性、可靠性、容錯性以及成本的考慮便發揮了作

    spark讀取kafka資料寫入hbase

    package com.prince.demo.test import java.util.UUID import com.typesafe.config.{Config, ConfigFactory} import org.apache.hadoop.hbase.HBa

    flume 讀取kafka 資料

    本文介紹flume讀取kafka資料的方法 程式碼： /*******************************************************************************  * Licensed to the Apache So

    《從0到1學習Flink》—— Flink 讀取 Kafka 資料寫入到 RabbitMQ

    開發十年，就只剩下這套架構體系了！ >>>   

    logstash寫入到kafka和從kafka讀取日誌

    使用 read ads oot put ssa tar json acc 收集nginx日誌放到kafka 修改nginx日誌格式：[nginx日誌格式修改](https://blog.51cto.com/9025736/2373483) input { file {

    flume 讀取kafka 數據

    any from 原來 string strac span value ive erro 本文介紹flume讀取kafka數據的方法 代碼： /*********

    IO流讀取資料檔案，將資料寫入資料庫，並記錄資料匯入日誌

    流程分析： 資料型別： ROUTE_ID,LXBM,ROAD_NAME,SRC_LON,SRC_LAT,DEST_LON,DEST_LAT 10000,G50,滬渝高速,115.8605349,30.08934467,115.5437817,30.08898601 10001,G

    在idea下使用java將Log4j日誌實時寫入Kafka（Kafka實時日誌寫入）

    本篇文章主要介紹在windows下使用idea新建web專案將Log4j日誌實時寫入Kafka。 簡要步驟： ①、新建web專案 ②、啟動zookeeper ③、啟動Kafka ④、建立topic ⑤、啟動 Kafka 消費者，執行專案，觀察Kafka 消費者的控制檯 &

    flume讀取binlog與kafka整合

    一、現將kafka調通   檢視zookeeper的topic cd /usr/software/zookeeper/zookeeper/bin ./zkCli.sh start ls /brokers/topics   先來說一下，刪除kafka