Hi, Harsh I configured flume on AWS to write in s3 bucket, while running flume agent it throwing some error My flume sink config: tier1.sinks.sink1.type = hdfs tier1.sinks.sink1.channel = channel1 tier1.sinks.sink1.hdfs.path = s3n://ACCESS_KEY_ID:SECRET_ACCESS_KEY@bucketname/ tier1.sinks.sink1.hdfs.filePrefix = Flumedata tier1.sinks.sink1.hdfs.fileType = DataStream tier1.sinks.sink1.hdfs.writeFormat= Text tier1.sinks.sink1.hdfs.batchSize = 100 tier1.sinks.sink1.hdfs.rollCount = 0 tier1.sinks.sink1.hdfs.rollSize = 73060835 tier1.sinks.sink1.hdfs.rollInterval = 0 #tier1.sinks.sink1.hdfs.idleTimeout = 180 #tier1.sinks.sink1.hdfs.closeTries = 0 and error: 2016-06-01 18:17:53,737 (SinkRunner-PollingRunner-DefaultSinkProcessor) [ERROR - org.apache.flume.sink.hdfs.HDFSEventSink.process(HDFSEventSink.java:459)] process failed java.lang.NoSuchMethodError: org.apache.http.impl.client.DefaultHttpClient.execute(Lorg/apache/http/client/methods/HttpUriRequest;)Lorg/apache/http/client/methods/CloseableHttpResponse; at amazon.emr.metrics.ClientUtil.getInstanceId(ClientUtil.java:115) at amazon.emr.metrics.MetricsConfig.getInstanceId(MetricsConfig.java:294) at amazon.emr.metrics.MetricsConfig.<init>(MetricsConfig.java:195) at amazon.emr.metrics.MetricsConfig.<init>(MetricsConfig.java:182) at amazon.emr.metrics.MetricsConfig.<init>(MetricsConfig.java:177) at amazon.emr.metrics.MetricsSaver.ensureSingleton(MetricsSaver.java:652) at amazon.emr.metrics.MetricsSaver.addInternal(MetricsSaver.java:332) at amazon.emr.metrics.MetricsSaver.addValue(MetricsSaver.java:178) at org.apache.hadoop.hdfs.DFSOutputStream.newStreamForCreate(DFSOutputStream.java:1667) at org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1692) at org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1627) at org.apache.hadoop.hdfs.DistributedFileSystem$7.doCall(DistributedFileSystem.java:448) at org.apache.hadoop.hdfs.DistributedFileSystem$7.doCall(DistributedFileSystem.java:444) at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81) at org.apache.hadoop.hdfs.DistributedFileSystem.create(DistributedFileSystem.java:444) at org.apache.hadoop.hdfs.DistributedFileSystem.create(DistributedFileSystem.java:387) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:913) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:894) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:791) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:780) at org.apache.flume.sink.hdfs.HDFSDataStream.doOpen(HDFSDataStream.java:86) at org.apache.flume.sink.hdfs.HDFSDataStream.open(HDFSDataStream.java:113) at org.apache.flume.sink.hdfs.BucketWriter$1.call(BucketWriter.java:246) at org.apache.flume.sink.hdfs.BucketWriter$1.call(BucketWriter.java:235) at org.apache.flume.sink.hdfs.BucketWriter$9$1.run(BucketWriter.java:679) at org.apache.flume.auth.SimpleAuthenticator.execute(SimpleAuthenticator.java:50) at org.apache.flume.sink.hdfs.BucketWriter$9.call(BucketWriter.java:676) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) Exception in thread "SinkRunner-PollingRunner-DefaultSinkProcessor" java.lang.NoSuchMethodError: org.apache.http.impl.client.DefaultHttpClient.execute(Lorg/apache/http/client/methods/HttpUriRequest;)Lorg/apache/http/client/methods/CloseableHttpResponse; at amazon.emr.metrics.ClientUtil.getInstanceId(ClientUtil.java:115) at amazon.emr.metrics.MetricsConfig.getInstanceId(MetricsConfig.java:294) at amazon.emr.metrics.MetricsConfig.<init>(MetricsConfig.java:195) at amazon.emr.metrics.MetricsConfig.<init>(MetricsConfig.java:182) at amazon.emr.metrics.MetricsConfig.<init>(MetricsConfig.java:177) at amazon.emr.metrics.MetricsSaver.ensureSingleton(MetricsSaver.java:652) at amazon.emr.metrics.MetricsSaver.addInternal(MetricsSaver.java:332) at amazon.emr.metrics.MetricsSaver.addValue(MetricsSaver.java:178) at org.apache.hadoop.hdfs.DFSOutputStream.newStreamForCreate(DFSOutputStream.java:1667) at org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1692) at org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1627) at org.apache.hadoop.hdfs.DistributedFileSystem$7.doCall(DistributedFileSystem.java:448) at org.apache.hadoop.hdfs.DistributedFileSystem$7.doCall(DistributedFileSystem.java:444) at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81) at org.apache.hadoop.hdfs.DistributedFileSystem.create(DistributedFileSystem.java:444) at org.apache.hadoop.hdfs.DistributedFileSystem.create(DistributedFileSystem.java:387) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:913) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:894) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:791) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:780) at org.apache.flume.sink.hdfs.HDFSDataStream.doOpen(HDFSDataStream.java:86) at org.apache.flume.sink.hdfs.HDFSDataStream.open(HDFSDataStream.java:113) at org.apache.flume.sink.hdfs.BucketWriter$1.call(BucketWriter.java:246) at org.apache.flume.sink.hdfs.BucketWriter$1.call(BucketWriter.java:235) at org.apache.flume.sink.hdfs.BucketWriter$9$1.run(BucketWriter.java:679) at org.apache.flume.auth.SimpleAuthenticator.execute(SimpleAuthenticator.java:50) at org.apache.flume.sink.hdfs.BucketWriter$9.call(BucketWriter.java:676) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) please help if you are familiar with this. Thanks in advance ... View more

Hi, Harsh Its working fine but i forgot to update. As per you said agent name is the problem in starting command now i changed it to tier1 so its working good. Thanks a lot for your reply. ... View more

hi Ben I tried like this as you said agent.sources.localsource.interceptors = search-replace agent.sources.localsource.interceptors.search-replace.type = search_replace # Remove leading alphanumeric characters in an event body. agent.sources.localsource.interceptors.search-replace.searchPattern = ###|## agent.sources.localsource.interceptors.search-replace.replaceString = | agent.sources.localsource.interceptors = regex-filter agent.sources.localsource.interceptors.regex-filter.type = regex_filter # Remove full event body. agent.sources.localsource.interceptors.regex-filter.searchPattern = "pagenotfound.php" agent.sources.localsource.interceptors.regex-filter.excludeEvents = true But flume not writing events at all if i remove regex_filter interceptor then its writing events so you have any idea how to use it . Thanks in advance. ... View more

Hi, I tried as you suggested i changed port num in hdfs path as hdfs://192.168.4.110:8020/user/hadoop/flumelogs/ but facing same issue. I'm thinking about flume sink is not writing fast due to network issue or less bandwidth is that correct?   and i have one more dout  i installed flume on data node is that causing any problem? Thanks in advance. ... View more

Hi , As you said i tried like this # Please paste flume.conf here. Example: # Sources, channels, and sinks are defined per # agent name, in this case 'tier1'. tier1.sources = source1 tier1.channels = channel1 tier1.sinks = sink1 sink2 sink3 tier1.sources.source1.type = avro tier1.sources.source1.bind= 192.168.4.110 tier1.sources.source1.port= 8021 tier1.sources.source1.channels = channel1 tier1.channels.channel1.type= memory tier1.sinks.sink1.type = hdfs tier1.sinks.sink1.channel = channel1 tier1.sinks.sink1.hdfs.path = hdfs://192.168.4.110:8021/user/hadoop/flumelogs/ tier1.sinks.sink1.hdfs.filePrefix = Flumedata tier1.sinks.sink1.hdfs.fileType = DataStream tier1.sinks.sink1.hdfs.writeFormat= Text tier1.sinks.sink1.hdfs.batchSize = 100 tier1.sinks.sink1.hdfs.rollCount = 0 tier1.sinks.sink1.hdfs.rollSize = 73060835 tier1.sinks.sink1.hdfs.rollInterval = 0 tier1.sinks.sink2.type = hdfs tier1.sinks.sink2.channel = channel1 tier1.sinks.sink2.hdfs.path = hdfs://192.168.4.110:8021/user/hadoop/flumelogs/ tier1.sinks.sink2.hdfs.filePrefix = Flumedata1 tier1.sinks.sink2.hdfs.fileType = DataStream tier1.sinks.sink2.hdfs.writeFormat= Text tier1.sinks.sink2.hdfs.batchSize = 100 tier1.sinks.sink2.hdfs.rollCount = 0 tier1.sinks.sink2.hdfs.rollSize = 73060835 tier1.sinks.sink2.hdfs.rollInterval = 0 tier1.sinks.sink3.type = hdfs tier1.sinks.sink3.channel = channel1 tier1.sinks.sink3.hdfs.path = hdfs://192.168.4.110:8021/user/hadoop/flumelogs/ tier1.sinks.sink3.hdfs.filePrefix = Flumedata2 tier1.sinks.sink3.hdfs.fileType = DataStream tier1.sinks.sink3.hdfs.writeFormat= Text tier1.sinks.sink3.hdfs.batchSize = 100 tier1.sinks.sink3.hdfs.rollCount = 0 tier1.sinks.sink3.hdfs.rollSize = 73060835 tier1.sinks.sink3.hdfs.rollInterval = 0 # Other properties are specific to each type of # source, channel, or sink. In this case, we # specify the capacity of the memory channel. tier1.channels.channel1.capacity = 10000 tier1.channels.channel1.transactionCapacity = 1000 Still facing same error. org.apache.flume.node.PollingPropertiesFileConfigurationProvider Configuration provider starting org.apache.flume.node.PollingPropertiesFileConfigurationProvider Reloading configuration file:/data/var/run/cloudera-scm-agent/process/1660-flume-AGENT/flume.conf org.apache.flume.conf.FlumeConfigurationProcessing:sink1 org.apache.flume.conf.FlumeConfigurationProcessing:sink3 org.apache.flume.conf.FlumeConfigurationProcessing:sink2 org.apache.flume.conf.FlumeConfigurationProcessing:sink3 org.apache.flume.conf.FlumeConfigurationProcessing:sink2 org.apache.flume.conf.FlumeConfigurationProcessing:sink2 org.apache.flume.conf.FlumeConfigurationProcessing:sink1 org.apache.flume.conf.FlumeConfigurationProcessing:sink2 org.apache.flume.conf.FlumeConfigurationProcessing:sink1 org.apache.flume.conf.FlumeConfiguration Added sinks: sink1 sink2 sink3 Agent: tier1 org.apache.flume.conf.FlumeConfigurationProcessing:sink1 org.apache.flume.conf.FlumeConfigurationProcessing:sink2 org.apache.flume.conf.FlumeConfigurationProcessing:sink2 org.apache.flume.conf.FlumeConfigurationProcessing:sink3 org.apache.flume.conf.FlumeConfigurationProcessing:sink3 org.apache.flume.conf.FlumeConfigurationProcessing:sink3 org.apache.flume.conf.FlumeConfigurationProcessing:sink2 org.apache.flume.conf.FlumeConfigurationProcessing:sink1 org.apache.flume.conf.FlumeConfigurationProcessing:sink1 org.apache.flume.conf.FlumeConfigurationProcessing:sink1 org.apache.flume.conf.FlumeConfigurationProcessing:sink2 org.apache.flume.conf.FlumeConfigurationProcessing:sink3 org.apache.flume.conf.FlumeConfigurationProcessing:sink2 org.apache.flume.conf.FlumeConfigurationProcessing:sink3 org.apache.flume.conf.FlumeConfigurationProcessing:sink3 org.apache.flume.conf.FlumeConfigurationProcessing:sink1 org.apache.flume.conf.FlumeConfigurationProcessing:sink3 org.apache.flume.conf.FlumeConfigurationProcessing:sink3 org.apache.flume.conf.FlumeConfigurationProcessing:sink2 org.apache.flume.conf.FlumeConfigurationProcessing:sink1 org.apache.flume.conf.FlumeConfigurationProcessing:sink1 org.apache.flume.conf.FlumeConfiguration Post-validation flume configuration contains configuration for agents: [tier1] org.apache.flume.node.AbstractConfigurationProviderCreating channels org.apache.flume.channel.DefaultChannelFactoryCreating instance of channel channel1 type memory org.apache.flume.node.AbstractConfigurationProviderCreated channel channel1 org.apache.flume.source.DefaultSourceFactoryCreating instance of source source1, type avro org.apache.flume.sink.DefaultSinkFactoryCreating instance of sink: sink1, type: hdfs org.apache.flume.sink.DefaultSinkFactoryCreating instance of sink: sink2, type: hdfs org.apache.flume.sink.DefaultSinkFactoryCreating instance of sink: sink3, type: hdfs org.apache.flume.node.AbstractConfigurationProvideChannel channel1 connected to [source1, sink1, sink2, sink3] org.apache.flume.node.Application Starting new configuration:{ sourceRunners:{source1=EventDrivenSourceRunner: { source:Avro source source1: { bindAddress: 192.168.4.110, port: 8021 } }} sinkRunners:{sink1=SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@672f151d counterGroup:{ name:null counters:{} } }, sink2=SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@441357d7 counterGroup:{ name:null counters:{} } }, sink3=SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@51ec072b counterGroup:{ name:null counters:{} } }} channels:{channel1=org.apache.flume.channel.MemoryChannel{name: channel1}} } org.apache.flume.node.ApplicationStarting Channel channel1 org.apache.flume.instrumentation.MonitoredCounterGroupMonitored counter group for type: CHANNEL, name: channel1: Successfully registered new MBean. org.apache.flume.instrumentation.MonitoredCounterGroup Component type: CHANNEL, name: channel1 started org.apache.flume.node.ApplicationStarting Sink sink1 org.apache.flume.node.ApplicationStarting Sink sink2 org.apache.flume.node.Application Starting Sink sink3 org.apache.flume.node.Application Starting Source source1 org.apache.flume.source.AvroSource Starting Avro source source1: { bindAddress: 192.168.4.110, port: 8021 }... org.apache.flume.instrumentation.MonitoredCounterGroup Monitored counter group for type: SINK, name: sink2: Successfully registered new MBean. org.apache.flume.instrumentation.MonitoredCounterGroup Monitored counter group for type: SINK, name: sink1: Successfully registered new MBean. org.apache.flume.instrumentation.MonitoredCounterGroup Component type: SINK, name: sink1 started org.apache.flume.instrumentation.MonitoredCounterGroup Component type: SINK, name: sink2 started org.apache.flume.instrumentation.MonitoredCounterGroup Monitored counter group for type: SINK, name: sink3: Successfully registered new MBean. org.apache.flume.instrumentation.MonitoredCounterGroup Component type: SINK, name: sink3 started org.mortbay.log Logging to org.slf4j.impl.Log4jLoggerAdapter(org.mortbay.log) via org.mortbay.log.Slf4jLog org.mortbay.log jetty-6.1.26.cloudera.4 org.mortbay.log Started SelectChannelConnector@0.0.0.0:41414 org.apache.flume.instrumentation.MonitoredCounterGroup Monitored counter group for type: SOURCE, name: source1: Successfully registered new MBean. org.apache.flume.instrumentation.MonitoredCounterGroup Component type: SOURCE, name: source1 started org.apache.flume.source.AvroSource Avro source source1 started. org.apache.avro.ipc.NettyServer [id: 0x077365c8, /192.168.6.118:60703 => /192.168.4.110:8021] OPEN org.apache.avro.ipc.NettyServer [id: 0x077365c8, /192.168.6.118:60703 => /192.168.4.110:8021] BOUND: /192.168.4.110:8021 org.apache.avro.ipc.NettyServer [id: 0x077365c8, /192.168.6.118:60703 => /192.168.4.110:8021] CONNECTED: /192.168.6.118:60703 org.apache.flume.sink.hdfs.HDFSDataStreamSerializer = TEXT, UseRawLocalFileSystem = false org.apache.flume.sink.hdfs.HDFSDataStreamSerializer = TEXT, UseRawLocalFileSystem = false org.apache.flume.sink.hdfs.HDFSDataStreamSerializer = TEXT, UseRawLocalFileSystem = false org.apache.flume.sink.hdfs.BucketWriter Creating hdfs://192.168.4.110:8021/user/hadoop/flumelogs//Flumedata.1463466791008.tmporg.apache.avro.ipc.NettyServer[id: 0x7a41116a, /192.168.4.110:38630 => /192.168.4.110:8021] OPEN org.apache.avro.ipc.NettyServer [id: 0x7a41116a, /192.168.4.110:38630 => /192.168.4.110:8021] BOUND: /192.168.4.110:8021 org.apache.avro.ipc.NettyServer [id: 0x7a41116a, /192.168.4.110:38630 => /192.168.4.110:8021] CONNECTED: /192.168.4.110:38630 org.apache.avro.ipc.NettyServer [id: 0x7a41116a, /192.168.4.110:38630 😆 /192.168.4.110:8021] DISCONNECTED org.apache.avro.ipc.NettyServer Unexpected exception from downstream. org.apache.avro.AvroRuntimeException: Excessively large list allocation request detected: 134352896 items! Connection closed. at org.apache.avro.ipc.NettyTransportCodec$NettyFrameDecoder.decodePackHeader(NettyTransportCodec.java:167) at org.apache.avro.ipc.NettyTransportCodec$NettyFrameDecoder.decode(NettyTransportCodec.java:139) at org.jboss.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:425) at org.jboss.netty.handler.codec.frame.FrameDecoder.cleanup(FrameDecoder.java:482) at org.jboss.netty.handler.codec.frame.FrameDecoder.channelDisconnected(FrameDecoder.java:365) at org.jboss.netty.channel.Channels.fireChannelDisconnected(Channels.java:396) at org.jboss.netty.channel.socket.nio.AbstractNioWorker.close(AbstractNioWorker.java:336) at org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink.handleAcceptedSocket(NioServerSocketPipelineSink.java:81) at org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink.eventSunk(NioServerSocketPipelineSink.java:36) at org.jboss.netty.handler.codec.oneone.OneToOneEncoder.handleDownstream(OneToOneEncoder.java:54) at org.jboss.netty.channel.Channels.close(Channels.java:812) at org.jboss.netty.channel.AbstractChannel.close(AbstractChannel.java:197) at org.apache.avro.ipc.NettyTransportCodec$NettyFrameDecoder.decodePackHeader(NettyTransportCodec.java:166) at org.apache.avro.ipc.NettyTransportCodec$NettyFrameDecoder.decode(NettyTransportCodec.java:139) at org.jboss.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:425) at org.jboss.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:310) at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:268) at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:255) at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:88) at org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:107) at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:312) at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:88) at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603) at java.lang.Thread.run(Thread.java:722) org.apache.flume.sink.hdfs.HDFSEventSink HDFS IO error java.io.EOFException: End of File Exception between local host is: "HadoopF02.hadoopslave1.com/192.168.4.110"; destination host is: "HadoopF02.hadoopslave1.com":8021; : java.io.EOFException; For more details see: http://wiki.apache.org/hadoop/EOFException at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57) at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) at java.lang.reflect.Constructor.newInstance(Constructor.java:525) at org.apache.hadoop.net.NetUtils.wrapWithMessage(NetUtils.java:791) at org.apache.hadoop.net.NetUtils.wrapException(NetUtils.java:764) at org.apache.hadoop.ipc.Client.call(Client.java:1472) at org.apache.hadoop.ipc.Client.call(Client.java:1399) at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:232) at $Proxy21.create(Unknown Source) at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.create(ClientNamenodeProtocolTranslatorPB.java:295) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:601) at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:187) at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102) at $Proxy22.create(Unknown Source) at org.apache.hadoop.hdfs.DFSOutputStream.newStreamForCreate(DFSOutputStream.java:1738) at org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1662) at org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1587) at org.apache.hadoop.hdfs.DistributedFileSystem$6.doCall(DistributedFileSystem.java:397) at org.apache.hadoop.hdfs.DistributedFileSystem$6.doCall(DistributedFileSystem.java:393) at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81) at org.apache.hadoop.hdfs.DistributedFileSystem.create(DistributedFileSystem.java:393) at org.apache.hadoop.hdfs.DistributedFileSystem.create(DistributedFileSystem.java:337) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:908) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:889) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:786) at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:775) at org.apache.flume.sink.hdfs.HDFSDataStream.doOpen(HDFSDataStream.java:86) at org.apache.flume.sink.hdfs.HDFSDataStream.open(HDFSDataStream.java:113) at org.apache.flume.sink.hdfs.BucketWriter$1.call(BucketWriter.java:246) at org.apache.flume.sink.hdfs.BucketWriter$1.call(BucketWriter.java:235) at org.apache.flume.sink.hdfs.BucketWriter$9$1.run(BucketWriter.java:679) at org.apache.flume.auth.SimpleAuthenticator.execute(SimpleAuthenticator.java:50) at org.apache.flume.sink.hdfs.BucketWriter$9.call(BucketWriter.java:676) at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:334) at java.util.concurrent.FutureTask.run(FutureTask.java:166) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603) at java.lang.Thread.run(Thread.java:722) Caused by: java.io.EOFException at java.io.DataInputStream.readInt(DataInputStream.java:392) at org.apache.hadoop.ipc.Client$Connection.receiveRpcResponse(Client.java:1071) at org.apache.hadoop.ipc.Client$Connection.run(Client.java:966) org.apache.flume.sink.hdfs.BucketWriter Creating hdfs://192.168.4.110:8021/user/hadoop/flumelogs//Flumedata1.1463466791008.tmp org.apache.avro.ipc.NettyServer [id: 0x7a41116a, /192.168.4.110:38630 😆 /192.168.4.110:8021] UNBOUND org.apache.avro.ipc.NettyServer [id: 0x7a41116a, /192.168.4.110:38630 😆 /192.168.4.110:8021] CLOSED org.apache.avro.ipc.NettyServer Connection to /192.168.4.110:38630 disconnected. org.apache.avro.ipc.NettyServer Unexpected exception from downstream. org.apache.avro.AvroRuntimeException: Excessively large list allocation request detected: 150994944 items! Connection closed. at org.apache.avro.ipc.NettyTransportCodec$NettyFrameDecoder.decodePackHeader(NettyTransportCodec.java:167) at org.apache.avro.ipc.NettyTransportCodec$NettyFrameDecoder.decode(NettyTransportCodec.java:139) at org.jboss.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:425) at org.jboss.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:310) at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:268) at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:255) at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:88) at org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:107) at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:312) at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:88) at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603) at java.lang.Thread.run(Thread.java:722) it continuing like this .   And this flume agent ruuning on data node is that any problem. Please help thanks in advance. ... View more

Hi, Thanks a lot for your reply As you said i want to define two more sinks like sink 1 and sink 2 and sink 3 for example this is my config file for sink: # Please paste flume.conf here. Example: # Sources, channels, and sinks are defined per # agent name, in this case 'tier1'. tier1.sources = source1 tier1.channels = channel1 tier1.sinks = sink1 tier1.sources.source1.type = avro tier1.sources.source1.bind= 192.168.4.110 tier1.sources.source1.port= 8021 tier1.sources.source1.channels = channel1 tier1.channels.channel1.type= memory tier1.sinks.sink1.type = hdfs tier1.sinks.sink1.channel = channel1 tier1.sinks.sink1.hdfs.path = hdfs://192.168.4.110:8021/user/hadoop/flumelogs/ tier1.sinks.sink1.hdfs.fileType = DataStream tier1.sinks.sink1.hdfs.writeFormat= Text tier1.sinks.sink1.hdfs.batchSize = 100 tier1.sinks.sink1.hdfs.rollCount = 0 tier1.sinks.sink1.hdfs.rollSize = 73060835 tier1.sinks.sink1.hdfs.rollInterval = 0 # Other properties are specific to each type of # source, channel, or sink. In this case, we # specify the capacity of the memory channel. tier1.channels.channel1.capacity = 10000 tier1.channels.channel1.transactionCapacity = 1000 so i want to add two more like this tier1.sinks.sink2.type = hdfs tier1.sinks.sink2.channel = channel1 tier1.sinks.sink2.hdfs.path = hdfs://192.168.4.110:8021/user/hadoop/flumelogs/ tier1.sinks.sink2.hdfs.fileType = DataStream tier1.sinks.sink2.hdfs.writeFormat= Text tier1.sinks.sink2.hdfs.batchSize = 100 tier1.sinks.sink2.hdfs.rollCount = 0 tier1.sinks.sink2.hdfs.rollSize = 73060835 tier1.sinks.sink2.hdfs.rollInterval = 0 tier1.sinks.sink3.type = hdfs tier1.sinks.sink3.channel = channel1 tier1.sinks.sink3.hdfs.path = hdfs://192.168.4.110:8021/user/hadoop/flumelogs/ tier1.sinks.sink3.hdfs.fileType = DataStream tier1.sinks.sink3.hdfs.writeFormat= Text tier1.sinks.sink3.hdfs.batchSize = 100 tier1.sinks.sink3.hdfs.rollCount = 0 tier1.sinks.sink3.hdfs.rollSize = 73060835 tier1.sinks.sink3.hdfs.rollInterval = 0 Thanks in advance. ... View more

Hi,       As you said i checked log first error        This is that error   Avro source source1: Unable to process event batch. Exception follows. org.apache.flume.ChannelException: Unable to put batch on required channel: org.apache.flume.channel.MemoryChannel{name: channel1} at org.apache.flume.channel.ChannelProcessor.processEventBatch(ChannelProcessor.java:200) at org.apache.flume.source.AvroSource.appendBatch(AvroSource.java:386) at sun.reflect.GeneratedMethodAccessor1.invoke(Unknown Source) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:601) at org.apache.avro.ipc.specific.SpecificResponder.respond(SpecificResponder.java:91) at org.apache.avro.ipc.Responder.respond(Responder.java:151) at org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.messageReceived(NettyServer.java:188) at org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) at org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:173) at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:560) at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:787) at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:296) at org.jboss.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:462) at org.jboss.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:443) at org.jboss.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:303) at org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:560) at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:555) at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:268) at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:255) at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:88) at org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:107) at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:312) at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:88) at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178) at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108) at org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603) at java.lang.Thread.run(Thread.java:722) Caused by: org.apache.flume.ChannelFullException: Space for commit to queue couldn't be acquired. Sinks are likely not keeping up with sources, or the buffer size is too tight at org.apache.flume.channel.MemoryChannel$MemoryTransaction.doCommit(MemoryChannel.java:130) at org.apache.flume.channel.BasicTransactionSemantics.commit(BasicTransactionSemantics.java:151) at org.apache.flume.channel.ChannelProcessor.processEventBatch(ChannelProcessor.java:192) ... 30 more   This  the full error and i checked that charts that channel reaching 100 % so what to do now for clearing this.     Thanks in advance.   ... View more