You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flume.apache.org by JP <jp...@gmail.com> on 2012/08/02 13:35:09 UTC
adding avro headers to the BodyTextEventSerializer failing
Hi,
Im trying avro source to be written to the hdfs.
my flume configurations:
Changes done my side in
BodyTextEventSerializer.java<https://mail.google.com/mail/?ui=2&ik=ca0e934e67&view=att&th=138e71aa19a944ce&attid=0.1&disp=safe&realattid=f_h5dr7kf40&zw>is
i have added
only headers to the write method:
the this code:
public void write(Event e) throws IOException {
Map<String, String> mapHeader = e.getHeaders();
Set<String> keySet =mapHeader.keySet();
for(String key: keySet){
String value = mapHeader.get(key);
String keyValue ="***** ------"+key+" : "+value;
out.write(keyValue.getBytes());
out.write('\n');
}
out.write(e.getBody());
out.write('\n');
}
*flume-conf.properties*
agent2.sources = seqGenSrc
agent2.channels = memoryChannel
agent2.sinks = loggerSink
agent2.sources.seqGenSrc.type = avrO
agent2.sources.seqGenSrc.bind=slcso-poc2-lnx
agent2.sources.seqGenSrc.port=41414
agent2.channels.memoryChannel.type = memory
agent2.channels.memoryChannel.capacity = 1000000
agent2.channels.memoryChannel.transactionCapacity = 1000000
agent2.channels.memoryChannel.keep-alive = 30
agent2.sources.seqGenSrc.channels = memoryChannel
agent2.sinks.loggerSink.type = hdfs
agent2.sinks.loggerSink.hdfs.path = hdfs://localhost:8020/data/
agent2.sinks.loggerSink.hdfs.fileType = DataStream
agent2.sinks.loggerSink.channel = memoryChannel
agent2.sinks.loggerSink.serializer =
org.apache.flume.serialization.CustomLogAvroEventSerializer
agent2.sinks.loggerSink.serializer.compressionCodec = snappy
agent2.sinks.loggerSink.serializer.syncIntervalBytes = 2048000
agent2.channels.memoryChannel.type = memory
*The exception im getting is:*
bash-3.2$ vi flume.log
2012-08-02 16:38:26,364 INFO hdfs.BucketWriter: Creating
hdfs://localhost:8020/data/FlumeData.1343905705610.tmp
2012-08-02 16:38:26,489 ERROR serialization.EventSerializerFactory: Unable
to instantiate Builder from
org.apache.flume.serialization.BodyTextEventSerializer
2012-08-02 16:38:26,490 WARN hdfs.HDFSEventSink: HDFS IO error
java.io.IOException: java.lang.NullPointerException
at
org.apache.flume.sink.hdfs.BucketWriter.doOpen(BucketWriter.java:202)
at
org.apache.flume.sink.hdfs.BucketWriter.access$000(BucketWriter.java:48)
at
org.apache.flume.sink.hdfs.BucketWriter$1.run(BucketWriter.java:155)
at
org.apache.flume.sink.hdfs.BucketWriter$1.run(BucketWriter.java:152)
at
org.apache.flume.sink.hdfs.BucketWriter.runPrivileged(BucketWriter.java:125)
at
org.apache.flume.sink.hdfs.BucketWriter.open(BucketWriter.java:152)
at
org.apache.flume.sink.hdfs.BucketWriter.append(BucketWriter.java:307)
at
org.apache.flume.sink.hdfs.HDFSEventSink$1.call(HDFSEventSink.java:717)
at
org.apache.flume.sink.hdfs.HDFSEventSink$1.call(HDFSEventSink.java:714)
at
java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
at java.util.concurrent.FutureTask.run(FutureTask.java:138)
at
java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)
Caused by: java.lang.NullPointerException
at
org.apache.flume.sink.hdfs.HDFSDataStream.open(HDFSDataStream.java:75)
at
org.apache.flume.sink.hdfs.BucketWriter.doOpen(BucketWriter.java:188)
... 13 more
2012-08-02 16:38:30,532 INFO hdfs.BucketWriter: Creating
hdfs://slcso-poc4-lnx:8020/data/cssplogs/FlumeData.1343905705611.tmp
2012-08-02 16:38:30,687 ERROR serialization.EventSerializerFactory: Unable
to instantiate Builder from
org.apache.flume.serialization.BodyTextEventSerializer
2012-08-02 16:38:30,724 WARN hdfs.HDFSEventSink: HDFS IO error
java.io.IOException: java.lang.NullPointerException
at
org.apache.flume.sink.hdfs.BucketWriter.doOpen(BucketWriter.java:202)
at
org.apache.flume.sink.hdfs.BucketWriter.access$000(BucketWriter.java:48)
at
org.apache.flume.sink.hdfs.BucketWriter$1.run(BucketWriter.java:155)
at
org.apache.flume.sink.hdfs.BucketWriter$1.run(BucketWriter.java:152)
at
org.apache.flume.sink.hdfs.BucketWriter.runPrivileged(BucketWriter.java:125)
at
org.apache.flume.sink.hdfs.BucketWriter.open(BucketWriter.java:152)
at
org.apache.flume.sink.hdfs.BucketWriter.append(BucketWriter.java:307)
at
org.apache.flume.sink.hdfs.HDFSEventSink$1.call(HDFSEventSink.java:717)
at
org.apache.flume.sink.hdfs.HDFSEventSink$1.call(HDFSEventSink.java:714)
at
java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
at java.util.concurrent.FutureTask.run(FutureTask.java:138)
at
java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)
Caused by: java.lang.NullPointerException
at
org.apache.flume.sink.hdfs.HDFSDataStream.open(HDFSDataStream.java:75)
at
org.apache.flume.sink.hdfs.BucketWriter.doOpen(BucketWriter.java:188)
... 13 more
2012-08-02 16:38:34,094 WARN ipc.NettyServer: Unexpected exception from
downstream.
java.io.IOException: Connection reset by peer
at sun.nio.ch.FileDispatcher.read0(Native Method)
at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:21)
at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:198)
at sun.nio.ch.IOUtil.read(IOUtil.java:166)
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:245)
at
org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:323)
at
org.jboss.netty.channel.socket.nio.NioWorker.processSelectedKeys(NioWorker.java:282)
at
org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:202)
at
java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)
2012-08-02 16:38:34,094 INFO ipc.NettyServer: [id: 0x722d95b3, /
10.77.235.245:3705 :> /10.105.39.202:41414] DISCONNECTED
2012-08-02 16:38:34,094 INFO ipc.NettyServer: [id: 0x722d95b3, /
10.77.235.245:3705 :> /10.105.39.202:41414] UNBOUND
2012-08-02 16:38:34,095 INFO ipc.NettyServer: [id: 0x722d95b3, /
10.77.235.245:3705 :> /10.105.39.202:41414] CLOSED
2012-08-02 16:38:35,750 INFO hdfs.BucketWriter: Creating
hdfs://slcso-poc4-lnx:8020/data/cssplogs/FlumeData.1343905705612.tmp
2012-08-02 16:38:35,784 ERROR serialization.EventSerializerFactory: Unable
to instantiate Builder from
org.apache.flume.serialization.BodyTextEventSerializer
2012-08-02 16:38:35,784 WARN hdfs.HDFSEventSink: HDFS IO error
java.io.IOException: java.lang.NullPointerException
at
org.apache.flume.sink.hdfs.BucketWriter.doOpen(BucketWriter.java:202)
at
org.apache.flume.sink.hdfs.BucketWriter.access$000(BucketWriter.java:48)
at
org.apache.flume.sink.hdfs.BucketWriter$1.run(BucketWriter.java:155)
at
org.apache.flume.sink.hdfs.BucketWriter$1.run(BucketWriter.java:152)
at
org.apache.flume.sink.hdfs.BucketWriter.runPrivileged(BucketWriter.java:125)
at
org.apache.flume.sink.hdfs.BucketWriter.open(BucketWriter.java:152)
at
org.apache.flume.sink.hdfs.BucketWriter.append(BucketWriter.java:307)
at
org.apache.flume.sink.hdfs.HDFSEventSink$1.call(HDFSEventSink.java:717)
at
org.apache.flume.sink.hdfs.HDFSEventSink$1.call(HDFSEventSink.java:714)
at
java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
at java.util.concurrent.FutureTask.run(FutureTask.java:138)
at
java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)
Caused by: java.lang.NullPointerException
at
org.apache.flume.sink.hdfs.HDFSDataStream.open(HDFSDataStream.java:75)
at
org.apache.flume.sink.hdfs.BucketWriter.doOpen(BucketWriter.java:188)
... 13 more
2012-08-02 16:38:40,808 INFO hdfs.BucketWriter: Creating
hdfs://slcso-poc4-lnx:8020/data/cssplogs/FlumeData.1343905705613.tmp
2012-08-02 16:38:40,914 ERROR serialization.EventSerializerFactory: Unable
to instantiate Builder from
org.apache.flume.serialization.BodyTextEventSerializer
2012-08-02 16:38:40,915 WARN hdfs.HDFSEventSink: HDFS IO error
java.io.IOException: java.lang.NullPointerException
at
org.apache.flume.sink.hdfs.BucketWriter.doOpen(BucketWriter.java:202)
at
org.apache.flume.sink.hdfs.BucketWriter.access$000(BucketWriter.java:48)
at
org.apache.flume.sink.hdfs.BucketWriter$1.run(BucketWriter.java:155)
at
org.apache.flume.sink.hdfs.BucketWriter$1.run(BucketWriter.java:152)
at
org.apache.flume.sink.hdfs.BucketWriter.runPrivileged(BucketWriter.java:125)
at
org.apache.flume.sink.hdfs.BucketWriter.open(BucketWriter.java:152)
at
org.apache.flume.sink.hdfs.BucketWriter.append(BucketWriter.java:307)
at
org.apache.flume.sink.hdfs.HDFSEventSink$1.call(HDFSEventSink.java:717)
at
org.apache.flume.sink.hdfs.HDFSEventSink$1.call(HDFSEventSink.java:714)
at
java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
at java.util.concurrent.FutureTask.run(FutureTask.java:138)
at
java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)
Caused by: java.lang.NullPointerException
--
JP