flume 데이터를elasticsearch 5로 보내기X 솔루션

agent.sources = r1
agent.sinks = es_sink
agent.channels = c1

# Describe/configure the source
agent.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.r1.zookeeperConnect = localhost:2181,localhost1:2181,
agent.sources.r1.topic =test
agent.sources.r1.groupId = flume
agent.sources.r1.kafka.consumer.timeout.ms = 100

# Describe the sink
agent.sinks.es_sink.type=com.cognitree.flume.sink.elasticsearch.ElasticSearchSink
agent.sinks.es_sink.es.bulkActions=5
agent.sinks.es_sink.es.bulkProcessor.name=bulkprocessor
agent.sinks.es_sink.es.bulkSize=5
agent.sinks.es_sink.es.bulkSize.unit=MB
agent.sinks.es_sink.es.concurrent.request=1
agent.sinks.es_sink.es.flush.interval.time=5m
agent.sinks.es_sink.es.backoff.policy.time.interval=50M
agent.sinks.es_sink.es.backoff.policy.retries=8
agent.sinks.es_sink.es.cluster.name=mycluster
agent.sinks.es_sink.es.client.transport.sniff=false
agent.sinks.es_sink.es.client.transport.ignore_cluster_name=false
agent.sinks.es_sink.es.client.transport.ping_timeout=5s
agent.sinks.es_sink.es.client.transport.nodes_sampler_interval=5s
agent.sinks.es_sink.es.client.hosts=localhost
agent.sinks.es_sink.es.client.port=9300
agent.sinks.es_sink.es.index=flume
agent.sinks.es_sink.es.type=flume-%y-%m-%d
agent.sinks.es_sink.es.index.builder=com.cognitree.flume.sink.elasticsearch.HeaderBasedIndexBuilder
agent.sinks.es_sink.es.serializer=com.cognitree.flume.sink.elasticsearch.SimpleSerializer
agent.sinks.es_sink.es.serializer.csv.fields=id:int,name:string,isemployee:boolean,leaves:float
agent.sinks.es_sink.es.serializer.csv.delimiter=,
agent.sinks.es_sink.es.serializer.avro.schema.file=/usr/local/schema.avsc


# Use a channel which buffers events in memory
agent.channels.c1.type = memory
agent.channels.c1.capacity = 100000
agent.channels.c1.transactionCapacity = 10000

# Bind the source and sink to the channel
agent.sources.r1.channels = c1
agent.sinks.es_sink.channel = c1