Flume使用大全之kafka source-kafka channel-hdfs

agent.sources = kafkaSource1 agent.channels = kafkaChannel agent.sinks = hdfsSink agent.sources.kafkaSource1.channels = kafkaChannel agent.sinks.hdfsSink.channel = kafkaChannel agent.sources.kafkaSource1.type = org.apache.flume.source.kafka.KafkaSource agent.sources.kafkaSource1.zookeeperConnect = node1:2181 agent.sources.kafkaSource1.topic = bpu_sensor_router,bpu_sensor_record_present,bpu_group_status_present,bpu_gateway_heartbeat,bpu_gateway_router,bpu_sensor_heartbeat agent.sources.kafkaSource1.consumer.group.id = flume agent.sources.kafkaSource1.kafka.consumer.timeout.ms = 100 agent.sources.kafkaSource1.kafka.bootstrap.servers = node7:9092 agent.sources.kafkaSource1.batchSize = 100 agent.sources.kafkaSource1.batchDurationMillis = 1000 agent.channels.kafkaChannel.type = org.apache.flume.channel.kafka.KafkaChannel agent.channels.kafkaChannel.kafka.bootstrap.servers = node7:9092 agent.channels.kafkaChannel.kafka.topic = flume-kafkaChannel agent.channels.kafkaChannel.consumer.group.id = flume-consumer #---------hdfsSink 相关配置------------------ agent.sinks.hdfsSink.type = hdfs # 注意, 我们输出到下面一个子文件夹data中 agent.sinks.hdfsSink.hdfs.path = hdfs://nameservice1/user/hive/warehouse/%{topic}/%Y/%m/%d agent.sinks.hdfsSink.hdfs.writeFormat = TEXT agent.sinks.hdfsSink.hdfs.fileType = DataStream agent.sinks.hdfsSink.hdfs.rollSize = 128000000 agent.sinks.hdfsSink.hdfs.rollInterval=60 agent.sinks.hdfsSink.hdfs.rollCount = 0 agent.sinks.hdfsSink.hdfs.batchSize = 100 agent.sinks.hdfsSink.hdfs.batchDurationMillis = 1000 agent.sinks.hdfsSink.hdfs.round = true agent.sinks.hdfsSink.hdfs.roundUnit = day agent.sinks.hdfsSink.hdfs.roundValue = 1 agent.sinks.hdfsSink.hdfs.threadsPoolSize = 25 agent.sinks.hdfsSink.hdfs.useLocalTimeStamp = true agent.sinks.hdfsSink.hdfs.minBlockReplicas = 1 agent.sinks.hdfsSink.hdfs.idleTimeout = 30 agent.sinks.hdfsSink.hdfs.filePrefix= %{topic}