1. a1.sources = r1
    2. a1.sinks = k1
    3. a1.channels = c1
    5. a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
    6. a1.sources.r1.kafka.bootstrap.servers = cdh-10-10-1-70.vm.crypto.lan:9092,cdh-10-10-1-71.vm.crypto.lan:9092,cdh-10-10-1-72.vm.crypto.lan:9092
    7. a1.sources.r1.kafka.topics=transactions
    8. # 调整这个参数性能提升很大，默认最大10000
    9. a1.sources.r1.batchSize=10000
    10. a1.sources.r1.batchDurationMillis=2000
    11. # 这个参数第一次使用的时候才有效，默认lastest,这个参数很重要，一定要配，不然换消费者了也还是消费最近的而不是最早的
    12. a1.sources.r1.kafka.consumer.auto.offset.reset = earliest
    13. a1.sources.r1.kafka.consumer.group.id=trans2
    15. # Describe the sink
    16. a1.sinks.k1.type = hdfs
    17. a1.sinks.k1.hdfs.path = hdfs://cdh-10-10-1-70.vm.crypto.lan:8020/flume/transactions/%Y%m%d/
    18. #上传文件的前缀
    19. a1.sinks.k1.hdfs.filePrefix = transactions-
    20. #是否按照时间滚动文件夹
    21. a1.sinks.k1.hdfs.round = true
    22. a1.sinks.k1.hdfs.roundValue = 1
    23. #重新定义时间单位
    24. a1.sinks.k1.hdfs.roundUnit = hour
    25. #是否使用本地时间戳
    26. a1.sinks.k1.hdfs.useLocalTimeStamp = true
    27. #积攒多少个Event才flush到HDFS一次
    28. a1.sinks.k1.hdfs.batchSize = 100
    29. #设置文件类型，可支持压缩
    30. a1.sinks.k1.hdfs.fileType = DataStream
    31. #多久生成一个新的文件
    32. a1.sinks.k1.hdfs.rollInterval = 3600
    33. #设置每个文件的滚动大小大概是128M
    34. a1.sinks.k1.hdfs.rollSize = 134217700
    35. #文件的滚动与Event数量无关
    36. a1.sinks.k1.hdfs.rollCount = 0
    38. # Use a channel which buffers events in memory
    39. a1.channels = c1
    40. a1.channels.c1.type = file
    41. a1.channels.c1.checkpointDir = /opt/flume/checkpoint
    42. a1.channels.c1.dataDirs = /opt/flume/data1,/opt/flume/data2
    44. # Bind the source and sink to the channel
    45. a1.sources.r1.channels = c1
    46. a1.sinks.k1.channel = c1