Kafka筆記整理（一）

[TOC]

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 
   1、為了減少磁盤寫入的次數,broker會將消息暫時buffer起來,當消息的個數達到一定閥值或者過了一定的時間間隔時,再flush到磁盤,這樣減少了磁盤IO調用的次數。
     配置：Log Flush Policy
     #log.flush.interval.messages=10000   一個分區的消息數閥值
     #log.flush.interval.ms=1000    
   2、kafka的消息保存一定時間（通常為7天）后會被刪除。
     配置：Log Retention Policy 
     log.retention.hours=168 
     #log.retention.bytes=1073741824
     log.retention.check.interval.ms=300000

Producer

Producer：配置文件：producer.properties
    1、自定義partition
    Producer也根據用戶設置的算法來根據消息的key來計算輸入哪個partition：partitioner.class
    2、異步或者同步發送
    配置項：producer.type
    異步或者同步發送
    同步是指：發送方發出數據后，等接收方發回響應以后才發下一個數據的通訊方式。  
    異步是指：發送方發出數據后，不等接收方發回響應，接著發送下個數據的通訊方式。
    3、批量發送可以很有效的提高發送效率。
    Kafka producer的異步發送模式允許進行批量發送，先將消息緩存在內存中，然后一次請求批量發送出去。
   具體配置queue.buffering.max.ms、queue.buffering.max.messages。
默認值分別為5000和10000

Consumer

consumers：配置文件：consumer.properties
1、每個consumer屬于一個consumer group，可以指定組id。group.id
2、消費形式：
   組內：組內的消費者消費同一份數據；同時只能有一個consumer消費一個Topic中的1個partition；一個consumer可以消費多個partitions中的消息。
     所以，對于一個topic,同一個group中推薦不能有多于partitions個數的consumer同時消費,否則將意味著某些consumer將無法得到消息。
   組間：每個消費組消費相同的數據，互不影響。
3、在一個consumer多個線程的情況下，一個線程相當于一個消費者。
   例如：partition為3，一個consumer起了3個線程消費，另一個后來的consumer就無法消費。

（這是Kafka用來實現一個Topic消息的廣播（發給所有的Consumer）和單播（發給某一個Consumer）的手段。
一個Topic可以對應多個Consumer Group。如果需要實現廣播，只要每個Consumer有一個獨立的Group就可以了。
要實現單播只要所有的Consumer在同一個Group里。用Consumer Group還可以將Consumer進行自由的分組而不需要多次發送消息到不同的Topic。）

topic、partition、message

1、每個partition在存儲層面是append log文件。新消息都會被直接追加到log文件的尾部，每條消息在log文件中的位置稱為offset（偏移量）。
2、每條Message包含了以下三個屬性：
  1°、offset 對應類型：long  此消息在一個partition中序號。可以認為offset是partition中Message的id
  2°、MessageSize  對應類型：int32 此消息的字節大小。
  3°、data  是message的具體內容。
3、越多的partitions意味著可以容納更多的consumer,有效提升并發消費的能力。
4、總之：業務區分增加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的配置文件也非常重要，有必要對其中的內容學習一下，這里給出其配置文件的說明：

# 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：

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 可以只寫一個，為了做筆記方便，后面只寫一個。

查看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：

不能修改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的相關概念，這里將uplooking01、uplooking02、uplooking03的brokerId分別修改為101、102、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的整合。