使用kafka作为生产者生产数据到hdfs

Posted 2020-11-09 瓶子xf

关键：查看kafka官网的userGuide

配置文件：

agent.sources = r1
agent.sinks = k1
agent.channels = c1

## sources config
agent.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.r1.kafka.bootstrap.servers = 192.168.80.128:9092,192.168.80.129:9092,192.168.80.130:9092
agent.sources.r1.kafka.topics =1711
agent.sources.r1.kafka.consumer.timeout.ms = 1000
agent.sources.r1.kafka.consumer.group.id = consumer-group111

## channels config
agent.channels.c1.type = memory
agent.channels.c1.capacity = 1000
agent.channels.c1.transactionCapacity = 100
agent.channels.c1.byteCapacityBufferPercentage = 60
agent.channels.c1.byteCapacity = 1280
agent.channels.c1.keep-alive = 60

# Describe the sink
agent.sinks.k1.type =hdfs
agent.sinks.k1.hdfs.path = hdfs://bcqm1711/kafkadir
agent.sinks.k1.hdfs.filePrefix = Syslog
agent.sinks.k1.hdfs.round = true
agent.sinks.k1.hdfs.roundValue = 1
agent.sinks.k1.hdfs.roundUnit = minute
agent.sinks.k1.hdfs.fileType=DataStream
agent.sinks.k1.hdfs.writeFormat=Text
agent.sinks.k1.hdfs.rollInterval=0
agent.sinks.k1.hdfs.rollSize=10240
agent.sinks.k1.hdfs.rollCount=0
agent.sinks.k1.hdfs.idleTimeout=60
agent.sinks.k1.hdfs.callTimeout=60000


# Bind the source and sink to the channel
agent.sources.r1.channels = c1
agent.sinks.k1.channel = c1