exec—Kafka

  1. a1.sources = r1
  2. a1.sinks = k1
  3. a1.channels = c1
  5. a1.sources.r1.type = exec
  6. a1.sources.r1.command = tail -F /home/admin/didi-car/output/order/order
  7. a1.sources.r1.fileHeader = true
  9. a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
  10. a1.sinks.k1.topic = hai_kou_order_topic
  11. a1.sinks.k1.brokerList= cdh01:9092,cdh02:9092,cdh03:9092
  12. a1.sinks.k1.batchSize = 20
  13. a1.sinks.k1.requiredAcks = 1
  14. a1.sinks.k1.producer.linger.ms = 1
  17. a1.channels.c1.type = memory
  18. a1.channels.c1.capacity = 1000
  19. a1.channels.c1.transactionCapacity = 100
  21. # Bind the source and sink to the channel
  22. a1.sources.r1.channels = c1
  23. a1.sinks.k1.channel = c1

TailDir—Kafka

一个source分发到两个channel中：

  1. a1.sources=r1
  2. a1.channels=c1 c2
  4. # configure source
  5. a1.sources.r1.type = TAILDIR
  6. a1.sources.r1.positionFile = /opt/module/flume/test/log_position.json
  7. a1.sources.r1.filegroups = f1
  8. a1.sources.r1.filegroups.f1 = /tmp/logs/app.+
  9. a1.sources.r1.fileHeader = true
  10. a1.sources.r1.channels = c1 c2
  12. #interceptor
  13. #com.atguigu.flume.interceptor.LogETLInterceptor和
  14. #com.atguigu.flume.interceptor.LogTypeInterceptor是自定义的拦截器的全类名。
  15. #需要根据用户自定义的拦截器做相应修改
  16. a1.sources.r1.interceptors =  i1 i2
  17. a1.sources.r1.interceptors.i1.type = com.atguigu.flume.interceptor.LogETLInterceptor$Builder
  18. a1.sources.r1.interceptors.i2.type = com.atguigu.flume.interceptor.LogTypeInterceptor$Builder
  20. a1.sources.r1.selector.type = multiplexing
  21. a1.sources.r1.selector.header = topic
  22. a1.sources.r1.selector.mapping.topic_start = c1
  23. a1.sources.r1.selector.mapping.topic_event = c2
  25. # configure channel
  26. a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
  27. a1.channels.c1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092
  28. a1.channels.c1.kafka.topic = topic_start
  29. a1.channels.c1.parseAsFlumeEvent = false
  30. a1.channels.c1.kafka.consumer.group.id = flume-consumer
  32. a1.channels.c2.type = org.apache.flume.channel.kafka.KafkaChannel
  33. a1.channels.c2.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092
  34. a1.channels.c2.kafka.topic = topic_event
  35. a1.channels.c2.parseAsFlumeEvent = false
  36. a1.channels.c2.kafka.consumer.group.id = flume-consumer

spooldir—-HDFS

  1. a1.sources = r1
  2. a1.sinks = k1
  3. a1.channels = c1
  5. ##注意：不能往监控目中重复丢同名文件
  6. a1.sources.r1.type = spooldir
  7. a1.sources.r1.spoolDir = /export/servers/dirfile
  8. a1.sources.r1.fileHeader = true
  10. # Describe the sink
  11. a1.sinks.k1.type = hdfs
  12. a1.sinks.k1.channel = c1
  13. a1.sinks.k1.hdfs.path = hdfs://node01:8020/spooldir/files/%y-%m-%d/%H%M/
  14. a1.sinks.k1.hdfs.filePrefix = events-
  15. a1.sinks.k1.hdfs.round = true
  16. a1.sinks.k1.hdfs.roundValue = 10
  17. a1.sinks.k1.hdfs.roundUnit = minute
  18. a1.sinks.k1.hdfs.rollInterval = 3
  19. a1.sinks.k1.hdfs.rollSize = 20
  20. a1.sinks.k1.hdfs.rollCount = 5
  21. a1.sinks.k1.hdfs.batchSize = 1
  22. a1.sinks.k1.hdfs.useLocalTimeStamp = true
  23. #生成的文件类型，默认是Sequencefile，可用DataStream，则为普通文本
  24. a1.sinks.k1.hdfs.fileType = DataStream
  26. # Use a channel which buffers events in memory
  27. a1.channels.c1.type = memory
  28. a1.channels.c1.capacity = 1000
  29. a1.channels.c1.transactionCapacity = 100
  31. # Bind the source and sink to the channel
  32. a1.sources.r1.channels = c1
  33. a1.sinks.k1.channel = c1