# The configuration file needs to define the sources,
# the channels and the sinks.
# Sources, channels and sinks are defined per agent,
# in this case called 'a
#agent section
producer.sources = s
producer.channels = c
producer.sinks = r
#producer.sources.s.type = seq
producer.sources.s.channels = c
#producer.sources.s.type = exec
#producer.sources.s.command=tail -n +0 -F /usr/local/nginx/nginxlog/access.log
producer.sources.s.deletePolicy=never
producer.sources.s.type = avro
producer.sources.s.bind = localhost
producer.sources.s.port = 4545
# Each sink's type must be defined(给谁了)
#producer.sinks.r.type = avro
#producer.sinks.r.hostname = 10.1.1.100
#producer.sinks.r.port = 20000
producer.sinks.r.type = org.xx.clickstream.sink.kafka.KafkaSink
producer.sinks.r.zk.connect = 127.0.0.1:2181
producer.sinks.r.metadata.broker.list=127.0.0.1:9092
producer.sinks.r.partitioner.class=org.xx.clickstream.partition.TypePartitioner
producer.sinks.r.serializer.class=kafka.serializer.StringEncoder
producer.sinks.r.request.required.acks=1
producer.sinks.r.max.message.size=1000000
producer.sinks.r.producer.type=sync
producer.sinks.r.custom.encoding=UTF-8
#Specify the channel the sink should use
producer.sinks.r.channel = c
# Each channel's type is defined.
producer.channels.c.type = memory
producer.channels.c.capacity = 1000000
producer.channels.c.transactionCapacity = 1000000
#producer.channels.c.type=file
#producer.channels.c.checkpointDir=/usr/local/flumeng/checkpointdir/tcpdir/example_agent
#producer.channels.c.dataDirs=/usr/local/flumeng/datadirs/tddirs/example_agen
sink配置(发送):
# The configuration file needs to define the sources,
# the channels and the sinks.
# Sources, channels and sinks are defined per agent,
# in this case called 'a
#agent section
producer.sources = s
producer.channels = c
producer.sinks = r
#producer.sources.s.type = seq
producer.sources.s.channels = c
producer.sources.s.type = exec
producer.sources.s.command=tail -n +0 -F /usr/local/nginx/nginxlog/access.log
producer.sources.s.deletePolicy=never
#producer.sources.s.type = avro
#producer.sources.s.bind = localhost
#producer.sources.s.port = 10000
# Each sink's type must be defined(给谁了)
producer.sinks.r.type = avro
producer.sinks.r.hostname = localhost
producer.sinks.r.port = 4545
#producer.sinks.r.type = org.xx.clickstream.sink.kafka.KafkaSink
#producer.sinks.r.zk.connect = 127.0.0.1:2181
#producer.sinks.r.metadata.broker.list=127.0.0.1:9092
#producer.sinks.r.partitioner.class=org.xx.clickstream.partition.TypePartitioner
#producer.sinks.r.serializer.class=kafka.serializer.StringEncoder
#producer.sinks.r.request.required.acks=1
producer.sinks.r.max.message.size=1000000
producer.sinks.r.producer.type=sync
producer.sinks.r.custom.encoding=UTF-8
#Specify the channel the sink should use
producer.sinks.r.channel = c
# Each channel's type is defined.
producer.channels.c.type = memory
producer.channels.c.capacity = 1000000
producer.channels.c.transactionCapacity = 1000000
#producer.channels.c.type=file
#producer.channels.c.checkpointDir=/usr/local/flumeng/checkpointdir/tcpdir/example_agent
#producer.channels.c.dataDirs=/usr/local/flumeng/datadirs/tddirs/example_agen
启动顺序,先启动source接收,再启动sink