file-flume-kafka.conf

  1. #为各组件命名
  2. a1.sources = r1
  3. a1.channels = c1
  5. #描述 
  6. source a1.sources.r1.type = TAILDIR
  7. a1.sources.r1.filegroups = f1
  8. a1.sources.r1.filegroups.f1 = /opt/module/applog/log/app.*
  9. a1.sources.r1.positionFile   = /opt/module/flume/taildir_position.json
  10. #自定义拦截器 校验json数据是否完整，可不加这个拦截器
  11. a1.sources.r1.interceptors = i1
  12. a1.sources.r1.interceptors.i1.type = com.atguigu.flume.interceptor.ETLInterceptor$Builder
  14. #描述 
  15. channel a1.channels.c1.type   = org.apache.flume.channel.kafka.KafkaChannel
  16. a1.channels.c1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092
  17. a1.channels.c1.kafka.topic = topic_log
  18. a1.channels.c1.parseAsFlumeEvent = false
  20. #绑定 source 和 channel 以及 sink 和 channel 的关系
  21. a1.sources.r1.channels = c1

kafka-flume-hdfs.conf

  1. ## 组件
  2. a1.sources=r1
  3. a1.channels=c1
  4. a1.sinks=k1
  6. ## source1
  7. a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource 
  8. a1.sources.r1.batchSize = 5000
  9. a1.sources.r1.batchDurationMillis = 2000
  10. a1.sources.r1.kafka.bootstrap.servers =hadoop102:9092,hadoop103:9092,hadoop104:9092
  11. #可以逗号分割topics 也可以正则匹配多个topics》》topic1，topic2   ^topic[0-9]$
  12. a1.sources.r1.kafka.topics=topic_log
  13. #自定义拦截器，修改header时间为数据时间，本拦截器必须添加 java代码详见尚硅谷文档 $Builder是固定写法，前边是全类名
  14. a1.sources.r1.interceptors = i1
  15. a1.sources.r1.interceptors.i1.type =com.atguigu.flume.interceptor.TimeStampInterceptor$Builder
  17. ## channel1
  18. a1.channels.c1.type = file
  19. a1.channels.c1.checkpointDir   = /opt/module/flume/checkpoint/behavior1
  20. a1.channels.c1.dataDirs = /opt/module/flume/data/behavior1/
  21. a1.channels.c1.maxFileSize = 2146435071
  22. a1.channels.c1.capacity = 1000000
  23. a1.channels.c1.keep-alive = 6
  25. ## sink1
  26. a1.sinks.k1.type = hdfs
  27. a1.sinks.k1.hdfs.path   = /origin_data/gmall/log/topic_log/%Y-%m-%d #输出格式
  28. a1.sinks.k1.hdfs.filePrefix = log-  #前缀
  29. a1.sinks.k1.hdfs.round = false
  30. #文件在128M时生成新文件、文件在1小时无写入后生成新文件
  31. a1.sinks.k1.hdfs.rollInterval = 10  #学习使用10，生产设置为3600 （3600秒一小时）
  32. a1.sinks.k1.hdfs.rollSize = 134217728
  33. a1.sinks.k1.hdfs.rollCount = 0
  35. ## 控制输出文件是lzop文件，此时输出的lzo文件没有索引 还需要后续手动创建
  36. a1.sinks.k1.hdfs.fileType = CompressedStream
  37. a1.sinks.k1.hdfs.codeC = lzop
  39. ## 拼装
  40. a1.sources.r1.channels = c1
  41. a1.sinks.k1.channel= c1