Kafka筆記整理（一）
[TOC]java
Kafka簡介
消息隊列（Message Queue）
消息 Message 網絡中的兩臺計算機或者兩個通信設備之間傳遞的數據。例如說：文本、音樂、視頻等內容。 隊列 Queue 一種特殊的線性表（數據元素首尾相接），特殊之處在於只容許在首部刪除元素和在尾部追加元素。入隊、出隊。 消息隊列 MQ 消息+隊列，保存消息的隊列。消息的傳輸過程當中的容器；主要提供生產、消費接口供外部調用作數據的存儲和獲取。
MQ分類
MQ主要分爲兩類：點對點(p2p)、發佈訂閱(Pub/Sub)
共同點：
消息生產者生產消息發送到queue中，而後消息消費者從queue中讀取而且消費消息。
不一樣點：
p2p模型包括：消息隊列(Queue)、發送者(Sender)、接收者(Receiver)
一個生產者生產的消息只有一個消費者(Consumer)(即一旦被消費，消息就不在消息隊列中)。好比說打電話。
Pub/Sub包含：消息隊列(Queue)、主題(Topic)、發佈者(Publisher)、訂閱者(Subscriber)
每一個消息能夠有多個消費者，彼此互不影響。好比我發佈一個微博：關注個人人都可以看到。
那麼在大數據領域呢，爲了知足日益增加的數據量，也有一款能夠知足百萬級別消息的生成和消費，分佈式、持久穩定的產品——Kafka。
Kafka簡介
Kafka是分佈式的發佈—訂閱消息系統。它最初由LinkedIn(領英)公司發佈，使用Scala語言編寫，與2010年12月份開源，成爲Apache的頂級項目。 Kafka是一個高吞吐量的、持久性的、分佈式發佈訂閱消息系統。 它主要用於處理活躍的數據(登陸、瀏覽、點擊、分享、喜歡等用戶行爲產生的數據)。 三大特色： 高吞吐量 能夠知足每秒百萬級別消息的生產和消費——生產消費。QPS 持久性 有一套完善的消息存儲機制，確保數據的高效安全的持久化——中間存儲。 分佈式 基於分佈式的擴展和容錯機制；Kafka的數據都會複製到幾臺服務器上。當某一臺故障失效時，生產者和消費者轉而使用其它的機器——總體健壯性。
Kafka組件
一個MQ須要哪些部分？生產、消費、消息類別、存儲等等。 對於kafka而言，kafka服務就像是一個大的水池。不斷的生產、存儲、消費着各類類別的消息。那麼kafka由何組成呢？ > Kafka服務： > Topic：主題，Kafka處理的消息的不一樣分類。 > Broker：消息代理，Kafka集羣中的一個kafka服務節點稱爲一個broker，主要存儲消息數據。存在硬盤中。每一個topic都是有分區的。 > Partition：Topic物理上的分組，一個topic在broker中被分爲1個或者多個partition，分區在建立topic的時候指定。 > Message：消息，是通訊的基本單位，每一個消息都屬於一個partition > Kafka服務相關 > Producer：消息和數據的生產者，向Kafka的一個topic發佈消息。 > Consumer：消息和數據的消費者，定於topic並處理其發佈的消息。 > Zookeeper：協調kafka的正常運行。
Broker
Broker：配置文件server.properties
一、爲了減小磁盤寫入的次數,broker會將消息暫時buffer起來,當消息的個數達到必定閥值或者過了必定的時間間隔時,再flush到磁盤,這樣減小了磁盤IO調用的次數。
配置：Log Flush Policy
#log.flush.interval.messages=10000 一個分區的消息數閥值
#log.flush.interval.ms=1000
二、kafka的消息保存必定時間（一般爲7天）後會被刪除。
配置：Log Retention Policy
log.retention.hours=168
#log.retention.bytes=1073741824
log.retention.check.interval.ms=300000
Producer
Producer：配置文件：producer.properties
一、自定義partition
Producer也根據用戶設置的算法來根據消息的key來計算輸入哪一個partition：partitioner.class
二、異步或者同步發送
配置項：producer.type
異步或者同步發送
同步是指：發送方發出數據後，等接收方發回響應之後才發下一個數據的通信方式。
異步是指：發送方發出數據後，不等接收方發回響應，接着發送下個數據的通信方式。
三、批量發送能夠頗有效的提升發送效率。
Kafka producer的異步發送模式容許進行批量發送，先將消息緩存在內存中，而後一次請求批量發送出去。
具體配置queue.buffering.max.ms、queue.buffering.max.messages。
默認值分別爲5000和10000
Consumer
consumers：配置文件：consumer.properties
一、每一個consumer屬於一個consumer group，能夠指定組id。group.id
二、消費形式：
組內：組內的消費者消費同一份數據；同時只能有一個consumer消費一個Topic中的1個partition；一個consumer能夠消費多個partitions中的消息。
因此，對於一個topic,同一個group中推薦不能有多於partitions個數的consumer同時消費,不然將意味着某些consumer將沒法獲得消息。
組間：每一個消費組消費相同的數據，互不影響。
三、在一個consumer多個線程的狀況下，一個線程至關於一個消費者。
例如：partition爲3，一個consumer起了3個線程消費，另外一個後來的consumer就沒法消費。
（這是Kafka用來實現一個Topic消息的廣播（發給全部的Consumer）和單播（發給某一個Consumer）的手段。
一個Topic能夠對應多個Consumer Group。若是須要實現廣播，只要每一個Consumer有一個獨立的Group就能夠了。
要實現單播只要全部的Consumer在同一個Group裏。用Consumer Group還能夠將Consumer進行自由的分組而不須要屢次發送消息到不一樣的Topic。）
topic、partition、message
一、每一個partition在存儲層面是append log文件。新消息都會被直接追加到log文件的尾部，每條消息在log文件中的位置稱爲offset（偏移量）。 二、每條Message包含了如下三個屬性： 1°、offset 對應類型：long 此消息在一個partition中序號。能夠認爲offset是partition中Message的id 2°、MessageSize 對應類型：int32 此消息的字節大小。 3°、data 是message的具體內容。 三、越多的partitions意味着能夠容納更多的consumer,有效提高併發消費的能力。 四、總之：業務區分增長topic、數據量大增長partition。
Kafka安裝配置
解壓： [uplooking@uplooking01 ~]$ tar -zxvf soft/kafka_2.10-0.10.0.1.tgz -C app/
重命名：[uplooking@uplooking01 ~]$ mv app/kafka_2.10-0.10.0.1/ app/kafka
添加KAFKA_HOME至環境變量：~/.bash_profile
export KAFKA_HOME=/home/uplooking/app/kafka
export PATH=$PATH:$KAFKA_HOME/bin
source ~/.bash_profile
配置相關參數：$KAFKA_HOME/config/server.properties
主要參數：broker.id、log.dirs、zookeeper.connect
broker.id=10
log.dirs=/home/uplooking/data/kafka [kafka數據的存放目錄]
zookeeper.connect=uplooking01:2181,uplooking02:2181,uplooking03:2181
kafka實例broker監聽默認端口9092，配置listeners=PLAINTEXT://:9092
啓動：
$KAFKA_HOME/bin/kafka-server-start.sh [-daemon] $KAFKA_HOME/config/server.properties
-daemon 可選，表示後臺啓動kafka服務
固然，kafka的配置文件也很是重要，有必要對其中的內容學習一下，這裏給出其配置文件的說明：node
# Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # see kafka.server.KafkaConfig for additional details and defaults ############################# Server Basics ############################# ################################################################################## # broker就是一個kafka的部署實例，在一個kafka集羣中，每一臺kafka都要有一個broker.id # 而且，該id惟一，且必須爲整數 ################################################################################## broker.id=10 ############################# Socket Server Settings ############################# # The address the socket server listens on. It will get the value returned from # java.net.InetAddress.getCanonicalHostName() if not configured. # FORMAT: # listeners = security_protocol://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 #listeners=PLAINTEXT://:9092 # Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for "listeners" if configured. Otherwise, it will use the value # returned from java.net.InetAddress.getCanonicalHostName(). #advertised.listeners=PLAINTEXT://your.host.name:9092 ################################################################################## #The number of threads handling network requests # 默認處理網絡請求的線程個數 3個 ################################################################################## num.network.threads=3 ################################################################################## # The number of threads doing disk I/O # 執行磁盤IO操做的默認線程個數 8 ################################################################################## num.io.threads=8 ################################################################################## # The send buffer (SO_SNDBUF) used by the socket server # socket服務使用的進行發送數據的緩衝區大小，默認100kb ################################################################################## socket.send.buffer.bytes=102400 ################################################################################## # The receive buffer (SO_SNDBUF) used by the socket server # socket服務使用的進行接受數據的緩衝區大小，默認100kb ################################################################################## socket.receive.buffer.bytes=102400 ################################################################################## # The maximum size of a request that the socket server will accept (protection against OOM) # socket服務所可以接受的最大的請求量，防止出現OOM(Out of memory)內存溢出，默認值爲：100m # （應該是socker server所能接受的一個請求的最大大小，默認爲100M） ################################################################################## socket.request.max.bytes=104857600 ############################# Log Basics （數據相關部分，kafka的數據稱爲log）############################# ################################################################################## # A comma seperated list of directories under which to store log files # 一個用逗號分隔的目錄列表，用於存儲kafka接受到的數據 ################################################################################## log.dirs=/home/uplooking/data/kafka ################################################################################## # The default number of log partitions per topic. More partitions allow greater # parallelism for consumption, but this will also result in more files across # the brokers. # 每個topic所對應的log的partition分區數目，默認1個。更多的partition數目會提升消費 # 並行度，可是也會致使在kafka集羣中有更多的文件進行傳輸 # （partition就是分佈式存儲，至關因而把一份數據分開幾份來進行存儲，即劃分塊、劃分分區的意思） ################################################################################## num.partitions=1 ################################################################################## # The number of threads per data directory to be used for log recovery at startup and flushing at shutdown. # This value is recommended to be increased for installations with data dirs located in RAID array. # 每個數據目錄用於在啓動kafka時恢復數據和在關閉時刷新數據的線程個數。若是kafka數據存儲在磁盤陣列中 # 建議此值能夠調整更大。 ################################################################################## num.recovery.threads.per.data.dir=1 ############################# Log Flush Policy （數據刷新策略）############################# # Messages are immediately written to the filesystem but by default we only fsync() to sync # the OS cache lazily. The following configurations control the flush of data to disk. # There are a few important trade-offs（平衡） here: # 1. Durability 持久性: Unflushed data may be lost if you are not using replication. # 2. Latency 延時性: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush. # 3. Throughput 吞吐量: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks. # The settings below allow one to configure the flush policy to flush data after a period of time or # every N messages (or both). This can be done globally and overridden on a per-topic basis. # kafka中只有基於消息條數和時間間隔數來制定數據刷新策略，而沒有大小的選項，這兩個選項能夠選擇配置一個 # 固然也能夠兩個都配置，默認狀況下兩個都配置，配置以下。 # The number of messages to accept before forcing a flush of data to disk # 消息刷新到磁盤中的消息條數閾值 #log.flush.interval.messages=10000 # The maximum amount of time a message can sit in a log before we force a flush # 消息刷新到磁盤生成一個log數據文件的時間間隔 #log.flush.interval.ms=1000 ############################# Log Retention Policy（數據保留策略） ############################# # The following configurations control the disposal（清理） of log segments（分片）. The policy can # be set to delete segments after a period of time, or after a given size has accumulated（累積）. # A segment will be deleted whenever（不管什麼時間） *either* of these criteria（標準） are met. Deletion always happens # from the end of the log. # 下面的配置用於控制數據片斷的清理，只要知足其中一個策略（基於時間或基於大小），分片就會被刪除 # The minimum age of a log file to be eligible for deletion # 基於時間的策略，刪除日誌數據的時間，默認保存7天 log.retention.hours=168 # A size-based retention policy for logs. Segments are pruned from the log as long as the remaining # segments don't drop below log.retention.bytes. 1G # 基於大小的策略，1G #log.retention.bytes=1073741824 # The maximum size of a log segment file. When this size is reached a new log segment will be created. # 數據分片策略 log.segment.bytes=1073741824 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies 5分鐘 # 每隔多長時間檢測數據是否達到刪除條件 log.retention.check.interval.ms=300000 ############################# Zookeeper ############################# # Zookeeper connection string (see zookeeper docs for details). # This is a comma separated host:port pairs, each corresponding to a zk # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002". # You can also append an optional chroot string to the urls to specify the # root directory for all kafka znodes. zookeeper.connect=uplooking01:2181,uplooking02:2181,uplooking03:2181 # Timeout in ms for connecting to zookeeper zookeeper.connection.timeout.ms=6000
另外須要注意的是，kafka啓動後，會在zookeeper中建立相關的節點：算法
[zk: localhost:2181(CONNECTED) 1] ls /
[controller, brokers, zookeeper, yarn-leader-election, hadoop-ha, admin, isr_change_notification, consumers, config, hbase]
[zk: localhost:2181(CONNECTED) 5] get /brokers/ids/10
{"jmx_port":-1,"timestamp":"1521936591128","endpoints":["PLAINTEXT://uplooking01:9092"],"host":"uplooking01","version":3,"port":9092}
cZxid = 0xa000000c1
ctime = Sun Mar 25 08:09:50 CST 2018
mZxid = 0xa000000c1
mtime = Sun Mar 25 08:09:50 CST 2018
pZxid = 0xa000000c1
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x6762543b71390005
dataLength = 133
numChildren = 0
Kafka操做
topic操做
建立Topic：express
kafka-topics.sh --create --topic hadoop --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181 --partitions 1 --replication-factor 1 kafka-topics.sh --create --topic hive --zookeeper uplooking01:2181 --partitions 1 --replication-factor 1 kafka-topics.sh --create --topic hbase --zookeeper uplooking01:2181 --partitions 3 --replication-factor 1 建立topic過程的問題，replication-factor個數不能超過broker的個數 bin/kafka-topics.sh --create --topic sqoop --zookeeper uplooking01:2181 --partitions 3 --replication-factor 3 Error while executing topic command : replication factor: 3 larger than available brokers: 1 另外，在建立topic後，能夠在/home/uplooking/data/kafka目錄查看到分區的目錄， 有多少個分區就會相應建立多少個目錄。
uplooking01:2181,uplooking02:2181,uplooking03:2181 能夠只寫一個，爲了作筆記方便，後面只寫一個。apache
查看Topic列表：緩存
kafka-topics.sh --list --zookeeper uplooking01:2181
查看某一個具體的Topic：安全
kafka-topics.sh --describe xxx --zookeeper uplooking01:2181
Topic:xxx PartitionCount:3 ReplicationFactor:1 Configs:
Topic: xxx Partition: 0 Leader: 10 Replicas: 10 Isr: 10
Topic: xxx Partition: 1 Leader: 10 Replicas: 10 Isr: 10
Topic: xxx Partition: 2 Leader: 10 Replicas: 10 Isr: 10
PartitionCount：topic對應的partition的個數
ReplicationFactor：topic對應的副本因子，說白就是副本個數
Partition：partition編號，從0開始遞增
Leader：當前partition起做用的breaker.id
Replicas: 當前副本數據坐在的breaker.id，是一個列表，排在最前面的起做用
Isr：當前kakfa集羣中可用的breaker.id列表
修改Topic：bash
不能修改replication-factor，以及只能對partition個數進行增長，不能減小 kafka-topics.sh --alter --topic hive --zookeeper uplooking01:2181 --partitions 3 partition由3變爲2的時，拋出的異常： ERROR kafka.admin.AdminOperationException: The number of partitions for a topic can only be increased
刪除Topic：服務器
kafka-topics.sh --delete --topic hbase --zookeeper uplooking01:2181 Topic hbase is marked for deletion. Note: This will have no impact if delete.topic.enable is not set to true. 完全刪除一個topic，須要在server.properties中配置delete.topic.enable=true，不然只是標記刪除 配置完成以後，須要重啓kafka服務。
生產者消費者案例
使用kafka提供的標準生產消費腳本。網絡
生產數據：
kafka-console-producer.sh --broker-list uplooking01:9092 --topic hadoop 生產數據的時候須要指定：當前數據流向哪一個broker，以及哪個topic
消費數據：
kafka-console-consumer.sh --topic hadoop --zookeeper uplooking01:2181 說明：該消費語句，只能獲取最新的數據，要想歷史數據，須要添加選項--from-beginning /kafka-console-consumer.sh --topic hadoop --zookeeper uplooking01:2181 --from-beginning 在消費數據的時候，只須要指定topic，以及topic的元數據信息便可(在ZK中存放)，因此這裏須要使用zk
消費者--黑名單(blacklist)和白名單(whitelist)選項：
--blacklist 後面跟須要過濾的topic的列表，使用","隔開，意思是除了列表中的topic以外，都能接收其它topic的數據 --whitelist 後面跟須要過濾的topic的列表，使用","隔開，意思是除了列表中的topic以外，都不能接收其它topic的數據 eg: kafka-console-consumer.sh --zookeeper uplooking01:2181 --from-beginning --blacklist hadoop,hive kafka-console-consumer.sh --zookeeper uplooking01:2181 --from-beginning --whitelist hadoop,flume
kafka分佈式環境搭建與概念驗證
kafka中沒有主從節點的概念，所以只須要將kafka安裝目錄拷貝到其它節點上便可，不過須要注意的是，須要修改brokerId爲惟一的：
scp -r /home/uplooking/app/kafka/ uplooking@uplooking02:/home/uplooking/app scp -r /home/uplooking/app/kafka/ uplooking@uplooking03:/home/uplooking/app
爲了方便後面理解kafka的相關概念，這裏將uplooking0一、uplooking0二、uplooking03的brokerId分別修改成10一、10二、103.
在三個節點上分別啓動kafka：
kafka-server-start.sh -daemon app/kafka/config/server.properties
建立一個topic：
kafka-topics.sh --create --topic hadoop --partitions 3 --replication-factor 3 --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181
查看該topic的詳細信息：
[uplooking@uplooking01 ~]$ kafka-topics.sh --describe hbase --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181
Topic:hadoop PartitionCount:3 ReplicationFactor:3 Configs:
Topic: hadoop Partition: 0 Leader: 101 Replicas: 101,102,103 Isr: 101,102,103
Topic: hadoop Partition: 1 Leader: 102 Replicas: 102,103,101 Isr: 102,103,101
Topic: hadoop Partition: 2 Leader: 103 Replicas: 103,101,102 Isr: 103,101,102
再查看前面的解釋：
PartitionCount：topic對應的partition的個數
ReplicationFactor：topic對應的副本因子，說白就是副本個數
Partition：partition編號，從0開始遞增
Leader：當前partition起做用的breaker.id
Replicas: 當前副本數據坐在的breaker.id，是一個列表，排在最前面的起做用
Isr：當前kakfa集羣中可用的breaker.id列表
這樣就很容易理解了。
這意味着，三個分區在三個節點上都有保存數據的，能夠分別在每一個節點上查看相關的分區數據信息：
[uplooking@uplooking01 kafka]$ ll 總用量 24 -rw-rw-r-- 1 uplooking uplooking 0 3月 25 19:33 cleaner-offset-checkpoint drwxrwxr-x 2 uplooking uplooking 4096 3月 25 19:33 hadoop-0 drwxrwxr-x 2 uplooking uplooking 4096 3月 25 19:33 hadoop-1 drwxrwxr-x 2 uplooking uplooking 4096 3月 25 19:33 hadoop-2 -rw-rw-r-- 1 uplooking uplooking 56 3月 25 19:33 meta.properties -rw-rw-r-- 1 uplooking uplooking 37 3月 25 19:48 recovery-point-offset-checkpoint -rw-rw-r-- 1 uplooking uplooking 37 3月 25 19:49 replication-offset-checkpoint
爲了進一步理解相關概念，能夠嘗試把uplooking01上的kafka關掉，而後再查看topic的詳細信息：
[uplooking@uplooking01 ~]$ kafka-topics.sh --describe hadoop --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181
Topic:hadoop PartitionCount:3 ReplicationFactor:3 Configs:
Topic: hadoop Partition: 0 Leader: 102 Replicas: 101,102,103 Isr: 102,103
Topic: hadoop Partition: 1 Leader: 102 Replicas: 102,103,101 Isr: 102,103
Topic: hadoop Partition: 2 Leader: 103 Replicas: 103,101,102 Isr: 103,102
而後我的分析以下：
前面提到：業務區分增長topic、數據量大增長partition。
因此partition分區，是爲了把數據分散來存放，這比如日誌須要天天分割同樣，也是避免單個存儲位置數據量過多。
顯然，至於咱們的每一個消息存放在哪一個分區，kafka自己是有機制去進行計算的：
int hashCode = Math.abs("ttt".hashCode());
int partition = hashCode % 50;
具體這裏就不進行討論了。
另外，由於設置了3個副本因子，因此3個分區的數據在3個節點上都會有保存，同時爲了起到負載均衡的做用，kafka
會爲每一個分區設置一個leader節點來專門進行該分區數據的相關操做。
如今再去看前面kafka組件的理論知識，就很容易理解了。
Kafka和Flume的整合
整合場景說明
如上圖所示，通常的，Kafka生產的數據，是由Flume的Sink提供的，這裏咱們須要用到Flume集羣， 經過Flume集羣將Agent的日誌收集分發到 Kafka（供實時計算處理）和HDFS（離線計算處理）。 這裏，咱們使用Flume做爲日誌收集系統，將收集到的數據輸送到Kafka中間件，以供Storm去實時消費計算，整個流程從各個Web節點 上， 經過Flume的Agent代理收集日誌，而後彙總到Flume集羣，在由Flume的Sink將日誌輸送到Kafka集羣，完成數據的生產流程。
整合案例
先建立一個topic：
[uplooking@uplooking01 ~]$ kafka-topics.sh --create --topic flume-kafka --partitions 3 --replication-factor 3 --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181
Created topic "flume-kafka".
[uplooking@uplooking01 ~]$ kafka-topics.sh --describe flume-kafka --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181
Topic:flume-kafka PartitionCount:3 ReplicationFactor:3 Configs:
Topic: flume-kafka Partition: 0 Leader: 101 Replicas: 101,102,103 Isr: 101,102,103
Topic: flume-kafka Partition: 1 Leader: 102 Replicas: 102,103,101 Isr: 102,103,101
Topic: flume-kafka Partition: 2 Leader: 103 Replicas: 103,101,102 Isr: 103,101,102
啓動kafka消費者：
[uplooking@uplooking01 ~]$ kafka-console-consumer.sh --topic flume-kafka --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181
Flume的配置文件，這裏爲監聽一個目錄下的文件變化：
######################################################### ## ##主要做用是監聽目錄中的新增文件，採集到數據以後，輸出到kafka ## 注意：Flume agent的運行，主要就是配置source channel sink ## 下面的a1就是agent的代號，source叫r1 channel叫c1 sink叫k1 ######################################################### a1.sources = r1 a1.sinks = k1 a1.channels = c1 #對於source的配置描述 監聽目錄中的新增文件 a1.sources.r1.type = spooldir a1.sources.r1.spoolDir = /home/uplooking/data/flume/source a1.sources.r1.fileHeader = true a1.sources.r1.fileHeaderKey = filepath a1.sources.r1.fileSuffix = .OK a1.sources.r1.deletePolicy = immediate #對於sink的配置描述 使用kafka作數據的消費 a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.topic = flume-kafka a1.sinks.k1.brokerList = uplooking01:9092,uplooking02:9092,uplooking03:9092 a1.sinks.k1.requiredAcks = 1 a1.sinks.k1.batchSize = 20 #對於channel的配置描述 使用內存緩衝區域作數據的臨時緩存 a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 #經過channel c1將source r1和sink k1關聯起來 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1
啓動Flume：
flume-ng agent --conf conf --name a1 --conf-file conf/flume-kafka.conf
向被偵聽目錄中添加hello文件，其內容以下：
hello he hello me hello you
添加後查看kafka消費者端的輸出：
[uplooking@uplooking01 ~]$ kafka-console-consumer.sh --topic flume-kafka --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181 hello he hello me hello you
這樣就完成了Kafka和Flume的整合。
相關文章
- 1. kafka學習筆記：知識點整理
- 2. kafka學習筆記整理01
- 3. SPARK 筆記 (八) sparkstreaming與kafka整合 (一)
- 4. Kafka筆記8（管理Kafka）
- 5. Kafka 學習筆記歸整
- 6. 【kafka】kafka學習筆記（一）
- 7. ios category 筆記整理（一）
- 8. HBase筆記整理（一）
- 9. Hive筆記整理（一）
- 10. 整理一波node筆記
- 更多相關文章...
- • ASP.NET Razor - 標記 - ASP.NET 教程
- • BASE原理與最終一致性 - NoSQL教程
- • Tomcat學習筆記（史上最全tomcat學習筆記）
- • Docker 清理命令
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. python的安裝和Hello，World編寫
- 2. 重磅解讀：K8s Cluster Autoscaler模塊及對應華爲雲插件Deep Dive
- 3. 鴻蒙學習筆記2（永不斷更）
- 4. static關鍵字 和構造代碼塊
- 5. JVM筆記
- 6. 無法啓動 C/C++ 語言服務器。IntelliSense 功能將被禁用。錯誤: Missing binary at c:\Users\MSI-NB\.vscode\extensions\ms-vsc
- 7. 【Hive】Hive返回碼狀態含義
- 8. Java樹形結構遞歸（以時間換空間）和非遞歸（以空間換時間）
- 9. 數據預處理---缺失值
- 10. 都要2021年了，現代C++有什麼值得我們學習的？
歡迎關注本站公眾號,獲取更多信息
相關文章