kafka和flume整合

Kafka作为source：

配置文件：

**#定义各个模块**
a1.sources = kafka 
a1.sinks = log
a1.channels = c1

#配置kafka source
#source的类型为kafkaSource
a1.sources.kafka.type = org.apache.flume.source.kafka.KafkaSource
#消费者连接的zk集群地址
a1.sources.kafka.zookeeperConnect = crxy155:2181,crxy156:2181,crxy162:2181
#消费者消费的topic，只能是一个。
a1.sources.kafka.topic = hello
#kafka的组id
a1.sources.kafka.groupId = flume
#kafka的消费者连接超时时间单位毫秒
a1.sources.kafka.kafka.consumer.timeout.ms = 3000


# 配置logger sink
a1.sinks.log.type = logger

# 配置 memory channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 绑定三种组件的关系
a1.sources.kafka.channels = c1
a1.sinks.log.channel = c1

Kafka作为sink：

############### 注意要查看下kafka的lib中的一些依赖包到flume的lib下####################
#定义各个模块
a1.sources = netcat 
a1.sinks = kfk
a1.channels = c1

#配置netcat source
a1.sources.netcat.type = netcat
a1.sources.netcat.bind = 0.0.0.0
a1.sources.netcat.port = 44444

# 配置 kafka sink
a1.sinks.kfk.type = org.apache.flume.sink.kafka.KafkaSink
#topic 如果header里有“topic”字段，会使用header里topic对应的值。
a1.sinks.kfk.topic = hello
a1.sinks.kfk.brokerList = crxy155:9092,crxy156:9092,crxy162:9092
#如果header里含有key这个header则会根据key进行分区。


# 配置 memory channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 绑定三种组件的关系
a1.sources.netcat.channels = c1
a1.sinks.kfk.channel = c1

Kafka作为channel：

**Kafka作为channel：三种方式：
With Flume source and sink
With Flume source and interceptor but no sink
With Flume sink, but no source**

#定义各个模块
a1.sources = netcat
a1.channels = kafka

#配置netcat source
a1.sources.netcat.type = netcat
a1.sources.netcat.bind = 0.0.0.0
a1.sources.netcat.port = 44444

# 配置 kafka channel
a1.channels.kafka.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.kafka.capacity = 10000
a1.channels.kafka.transactionCapacity = 1000
a1.channels.kafka.zookeeperConnect= crxy155:2181,crxy156:2181,crxy162:2181
a1.channels.kafka.brokerList=crxy155:9092,crxy156:9092,crxy162:9092
a1.channels.kafka.topic=hello
a1.channels.kafka.groupId=flume

# 绑定组件的关系
a1.sources.netcat.channels = kafka