Flume 配置模板

1. spooldir -> hdfs 的配置

  2. * Conf 配置
  4. agentDw.sources = SrcAccessLog
  5. agentDw.channels = ChAccesslog
  6. agentDw.sinks = SinkAccesslog
  8. # set SrcAccessLog
  10. # SrcSafeClickLog Source 配置
  11. agentDw.sources.SrcSafeClickLog.type = spooldir
  12. agentDw.sources.SrcSafeClickLog.spoolDir = /data/log/uba/access_log
  13. # 忽略文件正则
  14. agentDw.sources.SrcSafeClickLog.ignorePattern = ^(.)*\\.tmp$
  15. # 输入字符编码
  16. agentDw.sources.SrcSafeClickLog.inputCharset = UTF-8
  17. # 反序列化方式
  18. agentDw.sources.SrcSafeClickLog.deserializer = LINE
  19. # 一行最大字数
  20. agentDw.sources.SrcSafeClickLog.deserializer.maxLineLength = 204800
  21. agentDw.sources.SrcSafeClickLog.deserializer.outputCharset = UTF-8
  22. # 解码错误政策处理规则, FAIL(失效) || IGNORE(忽略)
  23. agentDw.sources.SrcSafeClickLog.decodeErrorPolicy = IGNORE
  24. # 完成删除文件 immediate | never
  25. agentDw.sources.SrcSafeClickLog.deletePolicy = immediate
  26. # 批处理条数
  27. agentDw.sources.SrcSafeClickLog.batchSize = 1000
  28. # 递归检测目录(必须开启)
  29. agentDw.sources.SrcSafeClickLog.recursiveDirectorySearch = true
  30. # 上传文件的绝对路径(必须开启)
  31. agentDw.sources.SrcSafeClickLog.fileHeader = true
  32. agentDw.sources.SrcSafeClickLog.fileHeaderKey = file
  33. # 上传的文件名(必须开启)
  34. agentDw.sources.SrcSafeClickLog.basenameHeader = true
  35. agentDw.sources.SrcSafeClickLog.basenameHeaderKey = basename
  36. agentDw.sources.SrcSafeClickLog.channels = ChAccesslog
  39. # set ChAccesslog
  40. # 保存类型
  41. agentDw.channels.ChAccesslog.type = file
  42. agentDw.channels.ChAccesslog.checkpointDir = /data/log/test/checkpoint
  43. agentDw.channels.ChAccesslog.dataDirs = /data/log/test/data
  44. # 设置最大线程数
  45. agentDw.channels.ChAccesslog.threads = 10
  48. # SinkAccesslog 设置
  49. agentDw.sinks.SinkAccesslog.type = hdfs
  50. agentDw.sinks.SinkAccesslog.channel = ChDwAccesslog
  51. agentDw.sinks.SinkAccesslog.hdfs.path = hdfs://uhadoop-ociicy-master2:8020/flume/test/access_log_%Y%m%d
  52. # hdfs 创建文件前缀
  53. agentDw.sinks.SinkAccesslog.hdfs.filePrefix = access_log
  54. # hdfs 创建文件后缀
  55. agentDw.sinks.SinkAccesslog.hdfs.fileSuffix = .log
  57. # 临时写入时的前缀  
  58. agentDw.sinks.SinkAccesslog.hdfs.inUsePrefix = .
  59. agentDw.sinks.SinkAccesslog.hdfs.inUseSuffix = .tmp
  61. agentDw.sinks.SinkAccesslog.hdfs.round = true
  62. agentDw.sinks.SinkAccesslog.hdfs.roundValue = 10
  63. # 下舍入值的单位 second, minute or hour.
  64. agentDw.sinks.SinkAccesslog.hdfs.roundUnit = minute
  66. # 复制块, 用于控制滚动大小
  67. agentDw.sinks.SinkAccesslog.hdfs.minBlockReplicas=1
  68. # 文件大小来触发滚动(字节), 0: 永远不触发
  69. agentDw.sinks.SinkAccesslog.hdfs.rollSize = 0
  70. # 文件条数来触发滚动(数量), 0:永远不触发
  71. agentDw.sinks.SinkAccesslog.hdfs.rollCount = 0
  72. # 滚动前等待的秒数(秒), 0:没有时间间隔, 每隔多少秒产生一个新文件, 案例为 60 喵
  73. agentDw.sinks.SinkAccesslog.hdfs.rollInterval = 60
  75. # 写入格式
  76. agentDw.sinks.SinkAccesslog.hdfs.writeFormat = Text
  77. # 文件格式 :  SequenceFile, DataStream(数据不会压缩输出文件) or CompressedStream(压缩输出,需要选择一个压缩/解码器)
  78. agentDw.sinks.SinkAccesslog.hdfs.fileType = DataStream
  80. # 批处理达到这个上限, 写到 HDFS
  81. agentDw.sinks.SinkAccesslog.hdfs.batchSize = 100
  82. # hdfs 打开、写、刷新、关闭的超时时间, 毫秒
  83. agentDw.sinks.SinkAccesslog.hdfs.callTimeout = 60000
  84. # 使用本地时间
  85. agentDw.sinks.SinkAccesslog.hdfs.useLocalTimeStamp = true

2. syslogtcp -> file 配置

  • 必须先启动 Flume, 开启 TCP|UDP 端口, 保证 Syslog 可以通过指定端口发送日志数据
  2. agentDw.sources = SrcUbaAppActionLog
  3. agentDw.channels = ChUbaAppActionLog
  4. agentDw.sinks = SinkUbaAppActionLog
  6. # UbaAppActionLog source 配置
  7. agentDw.sources.SrcUbaAppActionLog.type = syslogtcp
  8. agentDw.sources.SrcUbaAppActionLog.port = 10001
  9. agentDw.sources.SrcUbaAppActionLog.host = 0.0.0.0
  10. agentDw.sources.SrcUbaAppActionLog.channels = ChUbaAppActionLog
  12. # UbaAppActionLog channels 配置
  13. agentDw.channels.ChUbaAppActionLog.type = file
  14. agentDw.channels.ChUbaAppActionLog.checkpointDir = /var/log/flume/uba_app_action/checkpoint
  15. agentDw.channels.ChUbaAppActionLog.dataDirs = /var/log/flume/uba_app_action/data
  16. agentDw.channels.ChUbaAppActionLog.threads = 2
  18. # UbaAppActionLog sinks 配置
  19. agentDw.sinks.SinkUbaAppActionLog.channel = ChUbaAppActionLog
  20. agentDw.sinks.SinkUbaAppActionLog.type = thrift
  21. agentDw.sinks.SinkUbaAppActionLog.hostname = log1
  22. agentDw.sinks.SinkUbaAppActionLog.port = 18889

3. flume 多端口写入写出

  1. # Name the components on this agent
  2. a1.sources = r1 r2
  3. a1.sinks = k1 k2
  4. a1.channels = c1 c2
  6. # 定义两个 sources 分别来自 syslogtcp 的 44441 和 44442 端口
  7. a1.sources.r1.type = syslogtcp
  8. a1.sources.r1.bind = localhost
  9. a1.sources.r1.port = 44441
  11. a1.sources.r2.type = syslogtcp
  12. a1.sources.r2.bind = localhost
  13. a1.sources.r2.port = 44442
  15. # 定义两个 sinks 分别写入到 hdfs 中的不同目录下。
  16. a1.sinks.k1.type = hdfs
  17. a1.sinks.k1.hdfs.path = hdfs://localhost:9000/flume/events1/%y-%m-%d/
  18. a1.sinks.k1.hdfs.fileType=DataStream
  19. a1.sinks.k1.hdfs.writeFormat=Text
  20. a1.sinks.k1.hdfs.filePrefix = events-
  21. a1.sinks.k1.hdfs.rollCount= 0
  22. a1.sinks.k1.hdfs.rollSize= 0
  23. a1.sinks.k1.hdfs.rollInterval= 300
  24. a1.sinks.k1.hdfs.batchSize = 10000
  25. a1.sinks.k1.hdfs.useLocalTimeStamp = true
  27. a1.sinks.k2.type = hdfs
  28. a1.sinks.k2.hdfs.path = hdfs://localhost:9000/flume/events2/%y-%m-%d/
  29. a1.sinks.k2.hdfs.fileType=DataStream
  30. a1.sinks.k2.hdfs.writeFormat=Text
  31. a1.sinks.k2.hdfs.filePrefix = events-
  32. a1.sinks.k2.hdfs.rollCount= 0
  33. a1.sinks.k2.hdfs.rollSize= 0
  34. a1.sinks.k2.hdfs.rollInterval= 300
  35. a1.sinks.k2.hdfs.batchSize = 10000
  36. a1.sinks.k2.hdfs.useLocalTimeStamp = true
  39. # 定义两个 channels 因为需要两个 sinks 进行消费
  40. a1.channels.c1.type = memory
  41. a1.channels.c1.capacity = 1000
  42. a1.channels.c1.transactionCapacity = 100
  44. a1.channels.c2.type = memory
  45. a1.channels.c2.capacity = 1000
  46. a1.channels.c2.transactionCapacity = 100
  48. # Bind the source and sink to the channel
  49. a1.sources.r1.channels = c1
  50. a1.sources.r2.channels = c2
  52. a1.sinks.k1.channel = c1
  53. a1.sinks.k2.channel = c2

4. flume 一个 sources 多个 sinks

  • 这里举例 flume 同时写入到 hdfs 和 kafka 
  1. a2.sources = r1
  2. a2.sinks = k1 k2
  3. a2.channels = c1 c2
  5. # 定义数据源来自 spooldir
  6. a2.sources.r1.type = spooldir
  7. a2.sources.r1.channels = c1
  8. a2.sources.r1.spoolDir = ~/work/test/flume_source
  9. a2.sources.r1.fileHeader = true
  11. # 写入到 kafka 端口为 9092 server 中的 test topic 中
  12. a2.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
  13. a2.sinks.k1.channel = c1
  14. a2.sinks.k1.kafka.topic = test
  15. a2.sinks.k1.kafka.bootstrap.servers = localhost:9092
  16. a2.sinks.k1.kafka.flumeBatchSize = 20
  17. a2.sinks.k1.kafka.producer.acks = 1
  18. a2.sinks.k1.kafka.producer.linger.ms = 1
  19. #a2.sinks.ki.kafka.producer.compression.type = snappy
  21. # 写入到 hdfs 目录下
  22. a2.sinks.k2.type = hdfs
  23. a2.sinks.k2.channel = c2
  24. a2.sinks.k2.hdfs.path = hdfs://localhost:9000/flume/events/%y-%m-%d/%H%M/%S
  25. a2.sinks.k2.hdfs.fileType=DataStream
  26. a2.sinks.k2.hdfs.writeFormat=Text
  27. a2.sinks.k2.hdfs.filePrefix = events-
  28. a2.sinks.k2.hdfs.round = true
  29. a2.sinks.k2.hdfs.roundValue = 10
  30. a2.sinks.k2.hdfs.roundUnit = minute
  31. a2.sinks.k2.hdfs.useLocalTimeStamp = true
  33. # 两个 sink 就要对应 两个 channels
  34. a2.channels.c1.type = memory
  35. a2.channels.c1.capacity = 1000
  36. a2.channels.c1.transactionCapacity = 100
  38. a2.channels.c2.type = memory
  39. a2.channels.c2.capacity = 1000
  40. a2.channels.c2.transactionCapacity = 100
  41. # Bind the source and sink to the channel
  42. a2.sources.r1.channels = c1 c2

5. 负载均衡和故障转移

  2. # 配置需要处理的 srouce channels slinks
  3. agentDw.sources = SrcDwAccessLog
  4. agentDw.channels = ChDwAccesslog
  5. agentDw.sinks = SinkDwAccesslog1 SinkDwAccesslogKafka
  7. # 对所有的出口 slink 做 Load balancing Sink Processor 负载平衡处理器配置, 防止远端单点故障
  8. agentDw.sinkgroups = SinkGroupSinkDwAccesslog
  11. # --- DwAccessLog  配置 Start --- #
  13. # SrcDwAccessLog source 配置
  14. agentDw.sources.SrcDwAccessLog.type = syslogudp
  15. agentDw.sources.SrcDwAccessLog.port = 10004
  16. agentDw.sources.SrcDwAccessLog.host = 0.0.0.0
  17. agentDw.sources.SrcDwAccessLog.channels = ChDwAccesslog
  19. # SrcDwAccessLog Interceptors 配置
  20. agentDw.sources.SrcDwAccessLog.interceptors = in1 in2
  21. # SrcDwAccessLog Search and Replace Interceptor 配置
  22. agentDw.sources.SrcDwAccessLog.interceptors.in1.type = search_replace
  23. # 正则替换 ^[a-zA-Z_]+\:[ ]{1} 或者 ^lb_access\:[ ]{1}
  24. agentDw.sources.SrcDwAccessLog.interceptors.in1.searchPattern = ^[a-zA-Z_]+\:[ ]{1}
  25. agentDw.sources.SrcDwAccessLog.interceptors.in1.replaceString =
  26. agentDw.sources.SrcDwAccessLog.interceptors.in1.charset = UTF-8
  27. # SrcDwAccessLog Timestamp Interceptor 配置
  28. agentDw.sources.SrcDwAccessLog.interceptors.in2.type = timestamp
  29. agentDw.sources.SrcDwAccessLog.interceptors.in2.preserveExisting = true
  31. # ChDwAccesslog channels 配置
  32. agentDw.channels.ChDwAccesslog.type = file
  33. agentDw.channels.ChDwAccesslog.checkpointDir = /var/log/flume/dw_access_log/checkpoint
  34. agentDw.channels.ChDwAccesslog.dataDirs = /var/log/flume/dw_access_log/data
  35. agentDw.channels.ChDwAccesslog.capacity = 10000
  36. agentDw.channels.ChDwAccesslog.threads = 2
  38. # SinkDwAccesslog To File sinks 配置
  39. #agentDw.sinks.SinkDwAccesslog.channel = ChDwAccesslog
  40. #agentDw.sinks.SinkDwAccesslog.type = file_roll
  41. #agentDw.sinks.SinkDwAccesslog.sink.directory = /var/log/flume/dw_access_log/test
  43. # SinkDwAccesslogKafka To Kafka 配置
  44. #agentDw.sinks.SinkDwAccesslogKafka.channel = ChDwAccesslog
  45. #agentDw.sinks.SinkDwAccesslogKafka.type = org.apache.flume.sink.kafka.KafkaSink
  46. #agentDw.sinks.SinkDwAccesslogKafka.kafka.bootstrap.servers = bi4:9092
  47. #agentDw.sinks.SinkDwAccesslogKafka.kafka.topic = accessLogTest
  48. #agentDw.sinks.SinkDwAccesslogKafka.kafka.flumeBatchSize = 20
  49. # 被接受的值为0(从不等待确认)，1(只等待领导)，-1(等待所有副本)将其设置为-1
  50. #agentDw.sinks.SinkDwAccesslogKafka.kafka.producer.acks = 1
  51. #agentDw.sinks.SinkDwAccesslogKafka.kafka.producer.linger.ms = 1
  52. #agentDw.sinks.SinkDwAccesslogKafka.kafka.producer.compression.type = snappy
  54. # SinkDwAccesslog0 To thrift sinks 配置
  55. agentDw.sinks.SinkDwAccesslog0.channel = ChDwAccesslog
  56. agentDw.sinks.SinkDwAccesslog0.type = thrift
  57. agentDw.sinks.SinkDwAccesslog0.hostname = log0
  58. agentDw.sinks.SinkDwAccesslog0.port = 18889
  59. # 批量提交的个数
  60. agentDw.sinks.SinkDwAccesslog0.batch-size = 1000
  61. # 请求超时时间, 单位毫秒
  62. agentDw.sinks.SinkDwAccesslog0.request-timeout = 20000
  63. # 连接超时时间, 单位毫秒
  64. agentDw.sinks.SinkDwAccesslog0.connect-timeout = 3000
  65. # 重新连接 source 的时间, 单位秒, 用于后端负载均衡的轮询时间
  66. # 重接秒数, 如在故障转移模式时, 当前的 slinks 故障时间超过阈值, 就会转移到另外一个 slinks 处理
  67. agentDw.sinks.SinkDwAccesslog0.connection-reset-interval = 300
  69. # SinkDwAccesslog1 To thrift sinks 配置
  70. agentDw.sinks.SinkDwAccesslog1.channel = ChDwAccesslog
  71. agentDw.sinks.SinkDwAccesslog1.type = thrift
  72. agentDw.sinks.SinkDwAccesslog1.hostname = log1
  73. agentDw.sinks.SinkDwAccesslog1.port = 18889
  74. agentDw.sinks.SinkDwAccesslog1.batch-size = 1000
  75. agentDw.sinks.SinkDwAccesslog1.request-timeout = 20000
  76. agentDw.sinks.SinkDwAccesslog1.connect-timeout = 3000
  77. agentDw.sinks.SinkDwAccesslog1.connection-reset-interval = 300
  79. # SinkGroupSinkDwAccesslog 负载均衡
  80. agentDw.sinkgroups.SinkGroupSinkDwAccesslog.sinks = SinkDwAccesslog0 SinkDwAccesslog1
  81. agentDw.sinkgroups.SinkGroupSinkDwAccesslog.processor.type = load_balance
  82. # random(随机) 和 round_robin(轮询)
  83. agentDw.sinkgroups.SinkGroupSinkDwAccesslog.processor.selector = round_robin
  84. # 当某个sink不可用时，就会被加入黑名单列表中，一定时间之后再从黑名单中移除，继续被尝试。
  85. agentDw.sinkgroups.SinkGroupSinkDwAccesslog.processor.backoff = true
  86. # 黑名单的最长有效期, 单位毫秒(这里配置是: 1800 S)
  87. agentDw.sinkgroups.SinkGroupSinkDwAccesslog.processor.selector.maxTimeOut = 1800000  
  89. # SinkGroupSinkDwAccesslog 故障转义
  90. agentDw.sinkgroups.SinkGroupSinkDwAccesslog.sinks = SinkDwAccesslog0 SinkDwAccesslog1
  91. agentDw.sinkgroups.SinkGroupSinkDwAccesslog.processor.type = failover
  92. agentDw.sinkgroups.SinkGroupSinkDwAccesslog.processor.priority.SinkDwAccesslog0 = 1
  93. agentDw.sinkgroups.SinkGroupSinkDwAccesslog.processor.priority.SinkDwAccesslog1 = 100
  94. agentDw.sinkgroups.SinkGroupSinkDwAccesslog.processor.maxpenalty = 10000
  96. # --- DwAccessLog  配置 End --- #

6. hdfs sink 参数说明

  1. agentDw.sinks.SinkDwAccesslog1.type = hdfs
  2. agentDw.sinks.SinkDwAccesslog1.channel = ChDwAccesslog
  3. # 写入目录和文件规则
  4. agentDw.sinks.SinkDwAccesslog1.hdfs.path = hdfs://uhadoop-ociicy-master2:8020/flume/dw_access_log/dw_access_log_%Y%m%d
  5. agentDw.sinks.SinkDwAccesslog1.hdfs.filePrefix = dw_access_log
  6. agentDw.sinks.SinkDwAccesslog1.hdfs.fileSuffix = .log
  8. # 写入文件前缀规则
  9. agentDw.sinks.SinkDwAccesslog1.hdfs.inUsePrefix = .
  10. agentDw.sinks.SinkDwAccesslog1.hdfs.inUseSuffix = .tmp
  11. # hdfs 舍弃时间
  12. agentDw.sinks.SinkDwAccesslog1.hdfs.round = true
  13. # 时间上进行”舍弃”的单位，包含：second,minute,hour
  14. agentDw.sinks.SinkDwAccesslog1.hdfs.roundUnit = minute
  15. # 时间上进行“舍弃”的值, 2015-10-16 17:38:59 会被舍弃成  17:35, 5 分钟内的时间都被舍弃掉
  16. agentDw.sinks.SinkDwAccesslog1.hdfs.roundValue = 5
  18. # 复制块, 用于控制滚动大小
  19. agentDw.sinks.SinkDwAccesslog1.hdfs.minBlockReplicas=1
  20. # hdfs 间隔多长将临时文件重命名成最终目标文件, 并新打开一个临时文件来写入数据, 0 则表示不根据时间来滚动文件 (单位秒)
  21. agentDw.sinks.SinkDwAccesslog1.hdfs.rollInterval = 300
  22. # hdfs 临时文件达到 rollSize 值, 则滚动成目标文件, 0 则表示不根据临时文件大小来滚动文件(单位：bytes)
  23. agentDw.sinks.SinkDwAccesslog1.hdfs.rollSize = 0
  24. #events 数据达到该数量时候，将临时文件滚动成目标文件, 0 则表示不根据 events 数据来滚动文件
  25. agentDw.sinks.SinkDwAccesslog1.hdfs.rollCount = 0
  28. # 写入格式(必须 Text)
  29. agentDw.sinks.SinkDwAccesslog1.hdfs.writeFormat = Text
  31. # 不压缩
  32. # 文件格式 :  SequenceFile, DataStream(数据不会压缩输出文件) or CompressedStream(压缩 Stream)
  33. agentDw.sinks.SinkDwAccesslog1.hdfs.fileType = DataStream  
  35. # 设置压缩方式(当使用 CompressedStream 时,保存文件为压缩格式): gzip, bzip2, lzo, lzop, snappy
  36. agentDw.sinks.SinkAccesslog.hdfs.codeC = snappy
  39. # 这个拦截器写事件输出流的身体没有任何转换或修改, 事件标题将被忽略
  40. agentDw.sinks.SinkSafeClickLog.sink.serializer = text
  41. # 换行符追加到每个事件
  42. agentDw.sinks.SinkSafeClickLog.sink.serializer.appendNewline = true
  45. # 每个批次刷新到 HDFS上 的 events 数量
  46. agentDw.sinks.SinkDwAccesslog1.hdfs.batchSize = 10000
  47. # hdfs 打开、写、刷新、关闭的超时时间, 毫秒
  48. agentDw.sinks.SinkDwAccesslog1.hdfs.callTimeout = 60000
  49. # 当目前被打开的临时文件在该参数指定的时间（秒）内，没有任何数据写入，则将该临时文件关闭并重命名成目标文件
  50. agentDw.sinks.SinkDwAccesslog1.hdfs.idleTimeout = 0
  51. # 使用本地时间
  52. agentDw.sinks.SinkDwAccesslog1.hdfs.useLocalTimeStamp = true
  57. # batchsize < transactionCapacity || batchsize = transactionCapacity

Souce

  1. # Sources http
  2. agentDw.sources.srcHttp.type = http
  3. agentDw.sources.srcHttp.port = 10102
  4. agentDw.sources.srcHttp.bind = 0.0.0.0
  5. agentDw.sources.srcHttp.handler = com.dw.flume.source.http.HTTPCustomHandler
  6. agentDw.sources.srcHttp.threads = 8
  7. agentDw.sources.srcHttp.selector.type = replicating
  8. agentDw.sources.srcHttp.channels = chHdfs chKafka
  9. ## Interceptors
  10. agentDw.sources.srcHttp.interceptors = in1
  11. agentDw.sources.srcHttp.interceptors.in1.type = timestamp
  12. agentDw.sources.srcHttp.interceptors.in1.preserveExisting = true
  16. # Sources thrift
  17. agentDw.sources.srcThrift.type = thrift
  18. agentDw.sources.srcThrift.port = 10202
  19. agentDw.sources.srcThrift.bind = 0.0.0.0
  20. agentDw.sources.srcThrift.threads = 8
  21. agentDw.sources.srcThrift.selector.type = replicating
  22. agentDw.sources.srcThrift.channels = chHdfs chKafka
  23. ## Interceptors
  24. agentDw.sources.srcThrift.interceptors = in1
  25. agentDw.sources.srcThrift.interceptors.in1.type = timestamp
  26. agentDw.sources.srcThrift.interceptors.in1.preserveExisting = true
  30. # Sources avro
  31. agentDw.sources.srcAvro.type = avro
  32. agentDw.sources.srcAvro.port = 10302
  33. agentDw.sources.srcAvro.bind = 0.0.0.0
  34. agentDw.sources.srcAvro.threads = 8
  35. # 压缩算法, 对应 slink 也要配置 compression-type = deflate, 默认 none
  36. agentDw.sources.srcAvro.compression-type = deflate
  37. agentDw.sources.srcAvro.selector.type = replicating
  38. agentDw.sources.srcAvro.channels = chHdfs chKafka
  39. ## Interceptors
  40. agentDw.sources.srcAvro.interceptors = in1
  41. agentDw.sources.srcAvro.interceptors.in1.type = timestamp
  42. agentDw.sources.srcAvro.interceptors.in1.preserveExisting = true

Ch

  1. # Channels file
  2. agentDw.channels.ch.type = file
  3. agentDw.channels.ch.checkpointDir = /var/log/flume/Hdfs/checkpoint
  4. agentDw.channels.ch.dataDirs = /var/log/flume/Hdfs/data
  5. # channel 队列记录最大的 events 事件数量
  6. agentDw.channels.ch.capacity = 100000000
  7. # 最大文件的大小 128M
  8. agentDw.channels.ch.maxFileSize = 134217728
  9. # 最少需要多少空间 512M
  10. agentDw.channels.ch.minimumRequiredSpace = 524288000
  11. # 超时时间, channel 中没有数据最长等待时间
  12. agentDw.channels.ch.keep-alive = 3
  13. agentDw.channels.ch.threads = 8
  14. # 事物最大条数
  15. agentDw.channels.ch.transactionCapacity = 2000000

Slink

  1. # Sinks http
  2. agentDw.sinks.sinkHttp.type = http
  3. agentDw.sinks.sinkHttp.channel = ch
  4. agentDw.sinks.sinkHttp.endpoint = http://node1:10501
  5. agentDw.sinks.sinkHttp.connectTimeout = 2000
  6. agentDw.sinks.sinkHttp.requestTimeout = 2000
  7. agentDw.sinks.sinkHttp.contentTypeHeader = Content-Type:application/json;charset=UTF-8
  8. agentDw.sinks.sinkHttp.acceptHeader = Content-Type:application/json;charset=UTF-8
  9. agentDw.sinks.sinkHttp.defaultBackoff = true
  10. agentDw.sinks.sinkHttp.defaultRollback = true
  11. agentDw.sinks.sinkHttp.defaultIncrementMetrics = false
  12. agentDw.sinks.sinkHttp.backoff.200 = false
  13. agentDw.sinks.sinkHttp.rollback.200 = false
  14. agentDw.sinks.sinkHttp.incrementMetrics.200 = true
  15. agentDw.sinks.sinkHttp.serializer.compressionCodec = snappy
  19. # Sinks avro
  20. agentDw.sinks.sinkKafka.type = avro
  21. agentDw.sinks.sinkKafka.channel = ch
  22. agentDw.sinks.sinkKafka.hostname = log1
  23. agentDw.sinks.sinkKafka.port = 10302
  24. # 连接超时(ms)
  25. agentDw.sinks.sinkKafka.request-timeout = 20000
  26. # 请求超时(ms)
  27. agentDw.sinks.sinkKafka.connect-timeout = 20000
  28. # 复位连接间隔
  29. agentDw.sinks.sinkKafka.reset-connection-interval = 20000
  30. # 压缩算法, 对应 source 也要配置 compression-type = deflate, 默认 none
  31. agentDw.sinks.sinkKafka.compression-type = deflate
  32. # 压缩级别, 0: 不压缩压缩, 1 ~ 9: 数越高越压缩略越高, 默认 6
  33. agentDw.sinks.sinkKafka.compression-level = 6
  34. # 一次获取 N 个 Event 提交.  batchsize < transactionCapacity || batchsize = transactionCapacity
  35. agentDw.sinks.sinkKafka.batch-size = 100000
  39. # Sinks hdfs
  40. agentDw.sinks.sinkHdfs.type = hdfs
  41. agentDw.sinks.sinkHdfs.channel = ch
  42. agentDw.sinks.sinkHdfs.hdfs.path = hdfs://nameservice1/ods/safe_realtime_click_tmp/%Y%m%d/%H
  43. agentDw.sinks.sinkHdfs.hdfs.filePrefix = from_the_stream
  44. agentDw.sinks.sinkHdfs.hdfs.inUsePrefix = .
  45. agentDw.sinks.sinkHdfs.hdfs.inUseSuffix = .tmp
  46. agentDw.sinks.sinkHdfs.hdfs.round = true
  47. agentDw.sinks.sinkHdfs.hdfs.roundValue = 5
  48. agentDw.sinks.sinkHdfs.hdfs.roundUnit = minute
  49. agentDw.sinks.sinkHdfs.hdfs.minBlockReplicas = 1
  50. agentDw.sinks.sinkHdfs.hdfs.rollSize = 0
  51. agentDw.sinks.sinkHdfs.hdfs.rollCount = 0
  52. agentDw.sinks.sinkHdfs.hdfs.rollInterval = 600
  53. agentDw.sinks.sinkHdfs.hdfs.writeFormat = Text
  54. agentDw.sinks.sinkHdfs.hdfs.fileType = CompressedStream
  55. agentDw.sinks.sinkHdfs.hdfs.codeC = gzip
  56. agentDw.sinks.sinkHdfs.sink.serializer = text
  57. agentDw.sinks.sinkHdfs.sink.serializer.appendNewline = true
  58. agentDw.sinks.sinkHdfs.hdfs.callTimeout = 30000
  59. agentDw.sinks.sinkHdfs.hdfs.idleTimeout = 0
  60. agentDw.sinks.sinkHdfs.hdfs.useLocalTimeStamp = true
  61. # 一次获取 N 个 Event 提交.  batchsize < transactionCapacity || batchsize = transactionCapacity
  62. agentDw.sinks.sinkHdfs.hdfs.batchSize = 100000
  66. # Sinks kafka
  67. agentDw.sinks.sinkKafka.type = org.apache.flume.sink.kafka.KafkaSink
  68. agentDw.sinks.sinkKafka.channel = ch
  69. agentDw.sinks.sinkKafka.kafka.bootstrap.servers = node4:9092,node5:9092,node6:9092
  70. agentDw.sinks.sinkKafka.kafka.topic = kafka_topic
  71. # 为该通道中的所有事件指定一个Kafka分区ID, 默认情况下，如果此属性未设置，事件将由Kafka生产者的partition器分配
  72. # agentDw.sinks.sinkStreamKafka.defaultPartitionId
  73. # 有多少副本必须在其被认为成功写入之前确认一条消息。被接受的值为0(从不等待确认)性能最好，1(只等待领导)一般，-1(等待所有副本)最差但是不会丢数据
  74. agentDw.sinks.sinkKafka.kafka.producer.acks = 1
  75. agentDw.sinks.sinkKafka.kafka.producer.linger.ms = 1
  76. agentDw.sinks.sinkKafka.kafka.producer.compression.type = snappy
  77. # 一次获取 N 个 Event 提交.  batchsize < transactionCapacity || batchsize = transactionCapacity
  78. agentDw.sinks.sinkKafka.kafka.flumeBatchSize = 100000

Load

  1. # 负载均衡
  2. agentDw.sinkgroups.sinkGroupsCollector.sinks = sinkHttp sinkKafka sinkHdfs
  3. agentDw.sinkgroups.sinkGroupsCollector.processor.type = load_balance
  4. agentDw.sinkgroups.sinkGroupsCollector.processor.selector = round_robin
  5. agentDw.sinkgroups.sinkGroupsCollector.processor.backoff = true
  6. agentDw.sinkgroups.sinkGroupsCollector.processor.selector.maxTimeOut = 1800000