Thanks Yuki, have created a JIRA ticket https://issues.apache.org/jira/browse/CASSANDRA-8924
On Thu, Mar 5, 2015 at 10:34 AM, Yuki Morishita <mor.y...@gmail.com> wrote: > Thanks. > It looks like a bug. Can you create a ticket on JIRA? > > https://issues.apache.org/jira/browse/CASSANDRA > > On Thu, Mar 5, 2015 at 7:56 AM, Aby Kuruvilla > <aby.kuruvi...@envisagesystems.com> wrote: > > Hi Yuki > > > > Thanks for the reply! > > > > Here is the log from Cassandra server for the stream failure > > > > INFO [STREAM-INIT-/192.168.56.1:58578] 2015-03-04 09:20:23,816 > > StreamResultFuture.java:109 - [Stream > #98ba8730-c279-11e4-b8e9-55374d280508 > > ID#0] Creating new streaming plan for Bulk Load > > INFO [STREAM-INIT-/192.168.56.1:58578] 2015-03-04 09:20:23,816 > > StreamResultFuture.java:116 - [Stream > #98ba8730-c279-11e4-b8e9-55374d280508, > > ID#0] Received streaming plan for Bulk Load > > INFO [STREAM-INIT-/192.168.56.1:58579] 2015-03-04 09:20:23,819 > > StreamResultFuture.java:116 - [Stream > #98ba8730-c279-11e4-b8e9-55374d280508, > > ID#0] Received streaming plan for Bulk Load > > INFO [STREAM-IN-/127.0.0.1] 2015-03-04 09:20:23,822 > > StreamResultFuture.java:166 - [Stream > #98ba8730-c279-11e4-b8e9-55374d280508 > > ID#0] Prepare completed. Receiving 1 files(617874 bytes), sending 0 > files(0 > > bytes) > > WARN [STREAM-IN-/127.0.0.1] 2015-03-04 09:20:23,823 > StreamSession.java:597 > > - [Stream #98ba8730-c279-11e4-b8e9-55374d280508] Retrying for following > > error > > java.io.IOException: CF d6d35793-729c-3cab-bee0-84e971e48675 was dropped > > during streaming > > at > > > org.apache.cassandra.streaming.compress.CompressedStreamReader.read(CompressedStreamReader.java:71) > > ~[main/:na] > > at > > > org.apache.cassandra.streaming.messages.IncomingFileMessage$1.deserialize(IncomingFileMessage.java:48) > > [main/:na] > > at > > > org.apache.cassandra.streaming.messages.IncomingFileMessage$1.deserialize(IncomingFileMessage.java:38) > > [main/:na] > > at > > > org.apache.cassandra.streaming.messages.StreamMessage.deserialize(StreamMessage.java:55) > > [main/:na] > > at > > > org.apache.cassandra.streaming.ConnectionHandler$IncomingMessageHandler.run(ConnectionHandler.java:245) > > [main/:na] > > at java.lang.Thread.run(Thread.java:745) [na:1.7.0_65] > > ERROR [STREAM-IN-/127.0.0.1] 2015-03-04 09:20:23,828 > StreamSession.java:477 > > - [Stream #98ba8730-c279-11e4-b8e9-55374d280508] Streaming error occurred > > java.lang.IllegalArgumentException: Unknown type 0 > > at > > > org.apache.cassandra.streaming.messages.StreamMessage$Type.get(StreamMessage.java:89) > > ~[main/:na] > > at > > > org.apache.cassandra.streaming.messages.StreamMessage.deserialize(StreamMessage.java:54) > > ~[main/:na] > > at > > > org.apache.cassandra.streaming.ConnectionHandler$IncomingMessageHandler.run(ConnectionHandler.java:245) > > ~[main/:na] > > at java.lang.Thread.run(Thread.java:745) [na:1.7.0_65] > > INFO [STREAM-IN-/127.0.0.1] 2015-03-04 09:20:23,829 > > StreamResultFuture.java:180 - [Stream > #98ba8730-c279-11e4-b8e9-55374d280508] > > Session with /127.0.0.1 is complete > > WARN [STREAM-IN-/127.0.0.1] 2015-03-04 09:20:23,829 > > StreamResultFuture.java:207 - [Stream > #98ba8730-c279-11e4-b8e9-55374d280508] > > Stream failed > > > > > > > > On Wed, Mar 4, 2015 at 1:18 PM, Yuki Morishita <mor.y...@gmail.com> > wrote: > >> > >> Do you have corresponding error in the other side of the stream > >> (/192.168.56.11)? > >> > >> > >> On Wed, Mar 4, 2015 at 9:11 AM, Aby Kuruvilla > >> <aby.kuruvi...@envisagesystems.com> wrote: > >> > I am trying to use the CqlBulkOutputFormat in a Hadoop job to bulk > load > >> > data > >> > into Cassandra. Was not able to find any documentation of this new > >> > output > >> > format , but from looking through the code this uses CQLSSTableWriter > to > >> > write SSTable files to disk , which are then streamed to Cassandra > using > >> > SSTableLoader. On running the Hadoop job, I can see that the SSTable > >> > files > >> > do get generated but fails to stream the data out. I get the same > >> > exception > >> > when I try with Cassndra node on "localhost" as well as a remote > >> > Cassandra > >> > cluster. Also I get this exception on C* versions 2.1.1, 2.1.2 and > >> > 2.1.3. > >> > > >> > Relevant portion of logs and stack trace > >> > > >> > 09:20:23.207 [Thread-6] WARN org.apache.cassandra.utils.CLibrary - > JNA > >> > link > >> > failure, one or more native method will be unavailable. > >> > 09:20:23.208 [Thread-6] DEBUG org.apache.cassandra.utils.CLibrary - > JNA > >> > link > >> > failure details: Error looking up function 'posix_fadvise': > >> > dlsym(0x7fff6ab8a5e0, posix_fadvise): symbol not found > >> > 09:20:23.504 [Thread-6] DEBUG o.apache.cassandra.io.util.FileUtils - > >> > Renaming > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-tmp-ka-1-Filter.db > >> > to > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-ka-1-Filter.db > >> > 09:20:23.505 [Thread-6] DEBUG o.apache.cassandra.io.util.FileUtils - > >> > Renaming > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-tmp-ka-1-Digest.sha1 > >> > to > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-ka-1-Digest.sha1 > >> > 09:20:23.505 [Thread-6] DEBUG o.apache.cassandra.io.util.FileUtils - > >> > Renaming > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-tmp-ka-1-Statistics.db > >> > to > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-ka-1-Statistics.db > >> > 09:20:23.505 [Thread-6] DEBUG o.apache.cassandra.io.util.FileUtils - > >> > Renaming > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-tmp-ka-1-Index.db > >> > to > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-ka-1-Index.db > >> > 09:20:23.505 [Thread-6] DEBUG o.apache.cassandra.io.util.FileUtils - > >> > Renaming > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-tmp-ka-1-CompressionInfo.db > >> > to > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-ka-1-CompressionInfo.db > >> > 09:20:23.506 [Thread-6] DEBUG o.apache.cassandra.io.util.FileUtils - > >> > Renaming > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-tmp-ka-1-TOC.txt > >> > to > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-ka-1-TOC.txt > >> > 09:20:23.506 [Thread-6] DEBUG o.apache.cassandra.io.util.FileUtils - > >> > Renaming > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-tmp-ka-1-Data.db > >> > to > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-ka-1-Data.db > >> > 09:20:23.727 [Thread-2] DEBUG o.a.c.i.s.m.MetadataSerializer - Load > >> > metadata > >> > for > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-ka-1 > >> > 09:20:23.729 [Thread-2] INFO o.a.c.io.sstable.SSTableReader - Opening > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-ka-1 > >> > (617874 bytes) > >> > 09:20:23.780 [Thread-2] INFO o.a.c.streaming.StreamResultFuture - > >> > [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508] Executing streaming plan for > Bulk > >> > Load > >> > 09:20:23.781 [StreamConnectionEstablisher:1] INFO > >> > o.a.c.streaming.StreamSession - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508] Starting streaming to > >> > /192.168.56.11 > >> > 09:20:23.781 [StreamConnectionEstablisher:1] DEBUG > >> > o.a.c.streaming.ConnectionHandler - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508] Sending stream init for > incoming > >> > stream > >> > 09:20:23.792 [StreamConnectionEstablisher:1] DEBUG > >> > o.a.c.streaming.ConnectionHandler - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508] Sending stream init for > outgoing > >> > stream > >> > 09:20:23.794 [STREAM-OUT-/192.168.56.11] DEBUG > >> > o.a.c.streaming.ConnectionHandler - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508] Sending Prepare (0 requests, 1 > >> > files} > >> > 09:20:23.795 [StreamConnectionEstablisher:1] INFO > >> > o.a.c.streaming.StreamResultFuture - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508 ID#0] Prepare completed. > Receiving > >> > 0 > >> > files(0 bytes), sending 1 files(617874 bytes) > >> > 09:20:23.799 [StreamConnectionEstablisher:1] INFO > >> > o.a.c.streaming.StreamCoordinator - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508, ID#0] Beginning stream session > >> > with > >> > /192.168.56.11 > >> > 09:20:23.799 [STREAM-OUT-/192.168.56.11] DEBUG > >> > o.a.c.streaming.ConnectionHandler - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508] Sending File (Header (cfId: > >> > d6d35793-729c-3cab-bee0-84e971e48675, #0, version: ka, estimated keys: > >> > 3072, > >> > transfer size: 617874, compressed?: true, repairedAt: 0), file: > >> > > >> > > /var/folders/bb/c4416mx95xsbb11jx5g5zq15ddhhh4/T/dev/participant-262ce044-0a2d-48f4-9baa-ad4d626e743a/dev-participant-ka-1-Data.db) > >> > 09:20:23.808 [STREAM-IN-/192.168.56.11] DEBUG > >> > o.a.c.streaming.ConnectionHandler - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508] Received Retry > >> > (d6d35793-729c-3cab-bee0-84e971e48675, #0) > >> > 09:20:23.808 [STREAM-IN-/192.168.56.11] DEBUG > >> > o.a.c.streaming.ConnectionHandler - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508] Received Session Failed > >> > 09:20:23.809 [STREAM-IN-/192.168.56.11] DEBUG > >> > o.a.c.streaming.ConnectionHandler - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508] Closing stream connection > handler > >> > on > >> > /192.168.56.11 > >> > 09:20:23.811 [STREAM-IN-/192.168.56.11] INFO > >> > o.a.c.streaming.StreamResultFuture - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508] Session with /192.168.56.11 is > >> > complete > >> > 09:20:23.812 [STREAM-IN-/192.168.56.11] WARN > >> > o.a.c.streaming.StreamResultFuture - [Stream > >> > #98ba8730-c279-11e4-b8e9-55374d280508] Stream failed > >> > 09:20:23.815 [STREAM-OUT-/192.168.56.11] ERROR > >> > o.a.c.streaming.StreamSession > >> > - [Stream #98ba8730-c279-11e4-b8e9-55374d280508] Streaming error > >> > occurred > >> > java.io.IOException: Broken pipe > >> > at sun.nio.ch.FileDispatcherImpl.write0(Native Method) > >> > ~[na:1.7.0_51] > >> > at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:47) > >> > ~[na:1.7.0_51] > >> > at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93) > >> > ~[na:1.7.0_51] > >> > at sun.nio.ch.IOUtil.write(IOUtil.java:51) ~[na:1.7.0_51] > >> > at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:487) > >> > ~[na:1.7.0_51] > >> > at > >> > > >> > > sun.nio.ch.FileChannelImpl.transferToTrustedChannel(FileChannelImpl.java:473) > >> > ~[na:1.7.0_51] > >> > at sun.nio.ch.FileChannelImpl.transferTo(FileChannelImpl.java:569) > >> > ~[na:1.7.0_51] > >> > at > >> > > >> > > org.apache.cassandra.streaming.compress.CompressedStreamWriter.write(CompressedStreamWriter.java:74) > >> > ~[cassandra-all-2.1.2.jar:2.1.2] > >> > at > >> > > >> > > org.apache.cassandra.streaming.messages.OutgoingFileMessage$1.serialize(OutgoingFileMessage.java:56) > >> > ~[cassandra-all-2.1.2.jar:2.1.2] > >> > at > >> > > >> > > org.apache.cassandra.streaming.messages.OutgoingFileMessage$1.serialize(OutgoingFileMessage.java:40) > >> > ~[cassandra-all-2.1.2.jar:2.1.2] > >> > at > >> > > >> > > org.apache.cassandra.streaming.messages.StreamMessage.serialize(StreamMessage.java:45) > >> > ~[cassandra-all-2.1.2.jar:2.1.2] > >> > at > >> > > >> > > org.apache.cassandra.streaming.ConnectionHandler$OutgoingMessageHandler.sendMessage(ConnectionHandler.java:346) > >> > [cassandra-all-2.1.2.jar:2.1.2] > >> > at > >> > > >> > > org.apache.cassandra.streaming.ConnectionHandler$OutgoingMessageHandler.run(ConnectionHandler.java:318) > >> > [cassandra-all-2.1.2.jar:2.1.2] > >> > at java.lang.Thread.run(Thread.java:744) [na:1.7.0_51] > >> > 09:20:23.816 [STREAM-OUT-/192.168.56.11] ERROR > >> > o.a.c.streaming.StreamSession > >> > - [Stream #98ba8730-c279-11e4-b8e9-55374d280508] Streaming error > >> > occurred > >> > java.io.IOException: Broken pipe > >> > at sun.nio.ch.FileDispatcherImpl.write0(Native Method) > >> > ~[na:1.7.0_51] > >> > at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:47) > >> > ~[na:1.7.0_51] > >> > at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93) > >> > ~[na:1.7.0_51] > >> > at sun.nio.ch.IOUtil.write(IOUtil.java:65) ~[na:1.7.0_51] > >> > at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:487) > >> > ~[na:1.7.0_51] > >> > at > >> > > >> > > org.apache.cassandra.io.util.DataOutputStreamAndChannel.write(DataOutputStreamAndChannel.java:48) > >> > ~[cassandra-all-2.1.2.jar:2.1.2] > >> > at > >> > > >> > > org.apache.cassandra.streaming.messages.StreamMessage.serialize(StreamMessage.java:44) > >> > ~[cassandra-all-2.1.2.jar:2.1.2] > >> > at > >> > > >> > > org.apache.cassandra.streaming.ConnectionHandler$OutgoingMessageHandler.sendMessage(ConnectionHandler.java:346) > >> > [cassandra-all-2.1.2.jar:2.1.2] > >> > at > >> > > >> > > org.apache.cassandra.streaming.ConnectionHandler$OutgoingMessageHandler.run(ConnectionHandler.java:326) > >> > [cassandra-all-2.1.2.jar:2.1.2] > >> > at java.lang.Thread.run(Thread.java:744) [na:1.7.0_51] > >> > > >> > ---------------------------------------------------- > >> > > >> > Here is what I have tried > >> > > >> > > >> > Hadoop Driver > >> > > >> > public class CassandraBulkImporter extends Configured implements Tool{ > >> > > >> > ..... > >> > > >> > public static void main(String[] args) throws Exception { > >> > int exitCode = ToolRunner.run(new > >> > CassandraBulkImporter(), args); > >> > System.exit(exitCode); > >> > } > >> > > >> > @Override > >> > public int run(String[] arg0) throws Exception { > >> > ....... > >> > Job job = new Job(conf); > >> > ...... > >> > > >> > job.setOutputFormatClass(CqlBulkOutputFormat.class); > >> > > >> > > >> > ConfigHelper.setOutputInitialAddress(job.getConfiguration(), > >> > "192.168.56.11"); > >> > > >> > ConfigHelper.setOutputPartitioner(job.getConfiguration(), > >> > "Murmur3Partitioner"); > >> > > >> > ConfigHelper.setOutputRpcPort(job.getConfiguration(), > >> > "9160"); > >> > > >> > ConfigHelper.setOutputKeyspace(job.getConfiguration(), > >> > CASSANDRA_KEYSPACE_NAME); > >> > ConfigHelper.setOutputColumnFamily( > >> > job.getConfiguration(), > >> > CASSANDRA_KEYSPACE_NAME, > >> > CASSANDRA_TABLE_NAME > >> > ); > >> > //Set the properties for CqlBulkOutputFormat > >> > MultipleOutputs.addNamedOutput(job, > >> > CASSANDRA_TABLE_NAME, CqlBulkOutputFormat.class, Object.class, > >> > List.class); > >> > > >> > CqlBulkOutputFormat.setColumnFamilySchema(job.getConfiguration(), > >> > CASSANDRA_TABLE_NAME, "CREATE TABLE dev.participant(........)"); > >> > > >> > > >> > > CqlBulkOutputFormat.setColumnFamilyInsertStatement(job.getConfiguration(), > >> > CASSANDRA_TABLE_NAME, "INSERT into dev.participant(........) values > >> > (?,?,?,?,?) "); > >> > > >> > ..... > >> > } > >> > > >> > } > >> > > >> > Reducer Code > >> > > >> > public class ReducerToCassandra extends Reducer<Text, Text, Object, > >> > List<ByteBuffer>> { > >> > > >> > private MultipleOutputs multipleOutputs; > >> > > >> > @SuppressWarnings("unchecked") > >> > protected void setup(Context context) throws IOException, > >> > InterruptedException { > >> > multipleOutputs = new MultipleOutputs(context); > >> > } > >> > > >> > @Override > >> > public void reduce(Text id, Iterable<Text> pInfo, Context context) > >> > throws > >> > IOException, InterruptedException { > >> > .... > >> > List<ByteBuffer> bVariables = new ArrayList<ByteBuffer>(); > >> > > >> > ..... > >> > multipleOutputs.write(CASSANDRA_TABLE1, null, bVariables); > >> > > >> > } > >> > > >> > > >> > > >> > >> > >> > >> -- > >> Yuki Morishita > >> t:yukim (http://twitter.com/yukim) > > > > > > > > -- > Yuki Morishita > t:yukim (http://twitter.com/yukim) >
