|-|{"files":{"flume-ng-configure":{"env":"plain"}},"tag":"bigdata"}

flume-ng-configure

emitter.sources = access avro1 thrift1 nc1
emitter.channels = c1
emitter.sinks = s1 s2 s3
emitter.sinkgroups = g1


emitter.sources.access.channels = c1
emitter.sources.access.type = exec
emitter.sources.access.command = tail -F -n 0 --pid `ps -o ppid= $$` /tmp/access.log | sed -e "s/^/host=`hostname --fqdn` category=access:/"
emitter.sources.access.shell = /bin/sh -c
emitter.sources.access.restartThrottle = 5000
emitter.sources.access.restart = true
emitter.sources.access.logStdErr = true
emitter.sources.access.interceptors = i1 i2 i3
emitter.sources.access.interceptors.i1.type = timestamp
emitter.sources.access.interceptors.i2.type = host
emitter.sources.access.interceptors.i2.useIP = false
emitter.sources.access.interceptors.i3.type = static
emitter.sources.access.interceptors.i3.key = category
emitter.sources.access.interceptors.i3.value = access

emitter.sources.avro1.channels = c1
emitter.sources.avro1.type = avro
emitter.sources.avro1.bind = localhost
emitter.sources.avro1.port = 3000
emitter.sources.avro1.interceptors = i1 i2 i3
emitter.sources.avro1.interceptors.i1.type = timestamp
emitter.sources.avro1.interceptors.i2.type = host
emitter.sources.avro1.interceptors.i2.useIP = false
emitter.sources.avro1.interceptors.i3.type = static
emitter.sources.avro1.interceptors.i3.key = category
emitter.sources.avro1.interceptors.i3.value = default

emitter.sources.thrift1.channels = c1
emitter.sources.thrift1.type = thrift
emitter.sources.thrift1.bind = localhost
emitter.sources.thrift1.port = 3001
emitter.sources.thrift1.interceptors = i1 i2 i3
emitter.sources.thrift1.interceptors.i1.type = timestamp
emitter.sources.thrift1.interceptors.i2.type = host
emitter.sources.thrift1.interceptors.i2.useIP = false
emitter.sources.thrift1.interceptors.i3.type = static
emitter.sources.thrift1.interceptors.i3.key = category
emitter.sources.thrift1.interceptors.i3.value = default

emitter.sources.nc1.channels = c1
emitter.sources.nc1.type = netcat
emitter.sources.nc1.bind = localhost
emitter.sources.nc1.port = 3002
emitter.sources.nc1.max-line-length = 20480
emitter.sources.nc1.interceptors = i1 i2 i3
emitter.sources.nc1.interceptors.i1.type = timestamp
emitter.sources.nc1.interceptors.i2.type = host
emitter.sources.nc1.interceptors.i2.useIP = false
emitter.sources.nc1.interceptors.i3.type = static
emitter.sources.nc1.interceptors.i3.key = category
emitter.sources.nc1.interceptors.i3.value = default


emitter.channels.c1.type = file
emitter.channels.c1.checkpointDir = /tmp/flume/emitter-c1/checkpoint
emitter.channels.c1.useDualCheckpoints = true
emitter.channels.c1.backupCheckpointDir = /tmp/flume/emitter-c1/checkpointBackup
emitter.channels.c1.dataDirs = /tmp/flume/emitter-c1/data


emitter.sinks.s1.channel = c1
emitter.sinks.s1.type = avro
emitter.sinks.s1.hostname = collector1
emitter.sinks.s1.port = 3000
emitter.sinks.s1.batch-size = 100
#emitter.sinks.s1.reset-connection-interval = 600
emitter.sinks.s1.compression-type = deflate

emitter.sinks.s2.channel = c1
emitter.sinks.s2.type = avro
emitter.sinks.s2.hostname = collector2
emitter.sinks.s2.port = 3000
emitter.sinks.s2.batch-size = 100
#emitter.sinks.s2.reset-connection-interval = 600
emitter.sinks.s2.compression-type = deflate

emitter.sinks.s3.channel = c1
emitter.sinks.s3.type = avro
emitter.sinks.s3.hostname = collector3
emitter.sinks.s3.port = 3000
emitter.sinks.s3.batch-size = 100
#emitter.sinks.s3.reset-connection-interval = 600
emitter.sinks.s3.compression-type = deflate


emitter.sinkgroups.g1.sinks = s1 s2 s3
emitter.sinkgroups.g1.processor.type = load_balance
emitter.sinkgroups.g1.processor.backoff = true
emitter.sinkgroups.g1.processor.selector = round_robin