DataStax Support Forums » Topic: Problem JOINing data from Cassandra and Cfs

DataStax Support Forums » Topic: Problem JOINing data from Cassandra and Cfs http://www.datastax.com/support-forums/topic/problem-joining-data-from-cassandra-and-cfs Software, Support, and Training for Apache Cassandra en-US Sun, 26 May 2013 03:30:33 +0000 http://bbpress.org/?v=1.0.3 <![CDATA[Search]]> q http://www.datastax.com/support-forums/search.php Anonymous on "Problem JOINing data from Cassandra and Cfs" http://www.datastax.com/support-forums/topic/problem-joining-data-from-cassandra-and-cfs#post-1719 Thu, 19 Apr 2012 20:41:48 +0000 Anonymous 1719@http://www.datastax.com/support-forums/ I'm having a similar problem, and it seems that the exception only happens when performing a group or join using the field that is derived from the CF row key. here's the simplest failing example: if I load a CF and flatten to get multiple tuples for each row <code> p1 = LOAD 'cassandra://keyspace/CF' USING CassandraStorage() AS (key:chararray, columns: bag{T: tuple(property:chararray, value:chararray)}); p2 = group p1 by columns::property; dump p2; </code> Works just fine, but if I change the group by <code> p2 = group p1 by key; dump p2; </code> will fail with <blockquote> Type mismatch in key from map: expected org.apache.pig.impl.io.NullableText, recieved org.apache.pig.impl.io.NullableBytesWritable </blockquote> However, if I do a load, using the same schema, from a text file with lines of the form <blockquote> key {(c1,v1),(c2,v2),(c3,v3)} </blockquote> I can group (or join) by the key successfully. Any suggestions on how to proceed? Thanks, John larsen on "Problem JOINing data from Cassandra and Cfs" http://www.datastax.com/support-forums/topic/problem-joining-data-from-cassandra-and-cfs#post-545 Fri, 07 Oct 2011 14:00:06 +0000 larsen 545@http://www.datastax.com/support-forums/ I am having troubles using Pig + Cassandra (I deployed a staging node using DataStax Auto-Clustering AMI 2.0, ami-edc30384 on an EC2 m1.large instance) Here a description of what I'm doing, with snippets from the pig script I'm using: <code> register '/home/ubuntu/src/pig/contrib/piggybank/java/piggybank.jar'; register '/home/ubuntu/src/pygmalion/udf/target/pygmalion-1.1.0-SNAPSHOT.jar'; define FromCassandraBag org.pygmalion.udf.FromCassandraBag(); define ToCassandraBag org.pygmalion.udf.ToCassandraBag(); data = LOAD 'cassandra://RawLogTest/RawLog2' using CassandraStorage() AS (key, columns: {T: tuple( name, value ) }); rows = FOREACH data GENERATE key, FLATTEN(FromCassandraBag('..., some_id, ...', columns)) AS ( ... some_id:int, ... ); rows_stripped = FOREACH rows { t = REGEX_EXTRACT( key, '([0-9]+)\\.([0-9]+)\\.([0-9]+)', 1 ); GENERATE (long)t AS timestamp, ..., some_id, ...; } </code> Now I select a slice of rows_stripped, using parameters coming from the user. Then I GROUP and generate a report from the raw data. <code> interval = FILTER rows_stripped BY (timestamp >= $FROM and timestamp <= $TO); grouped = GROUP rows_stripped BY (...); report = FOREACH grouped GENERATE FLATTEN( group ), COUNT( rows_stripped ); </code> So far so good. I can DUMP all the relations defined, getting back the data I expect. Now I gather some other data from CSV files I previously put on Cfs (they actually come from a db table, which I imported via Sqoop). My purpose is to join the two relations to augment the report. <code> other_info = LOAD 'lat.csv' using PigStorage(',') as (... , some_id:int, ...); </code> This relation is DUMPable as well without problems. And now, I join the data with this.augmented_report: <code> augmented_report = JOIN other_info BY some_id, report BY some_id; </code> When I try to STORE or DUMP augmented_report, the mapreduce fails with the following error: <code> java.io.IOException: Type mismatch in key from map: expected org.apache.pig.impl.io.NullableIntWritable, recieved org.apache.pig.impl.io.NullableBytesWritable at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.collect(MapTask.java:1013) at org.apache.hadoop.mapred.MapTask$NewOutputCollector.write(MapTask.java:690) at org.apache.hadoop.mapreduce.TaskInputOutputContext.write(TaskInputOutputContext.java:80) at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Map.collect(PigMapReduce.java:116) at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapBase.runPipeline(PigMapBase.java:239) at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapBase.map(PigMapBase.java:232) at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapBase.map(PigMapBase.java:53) at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:144) at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:763) at org.apache.hadoop.mapred.MapTask.run(MapTask.java:369) at org.apache.hadoop.mapred.Child$4.run(Child.java:259) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:396) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1059) at org.apache.hadoop.mapred.Child.main(Child.java:253) </code> I can solve this problem storing report in Cfs, then JOINing other_info against a new relation, but I'd like to be able to do my augmented_report using data from the original source. What strategy do you recommend to debug and solve the problem ? thank you, s.