我正在处理一个数据流,包括Pig中的一些聚合步骤和将步骤存储到Cassandra中。我已经能够传递相对简单的数据类型,如integer、long或date,但找不到如何使用CqlStorage将某种列表、集合或元组从Pig传递给Cassandra。
我使用Pig 0.9.2,所以我不能使用FLATTEN方法。
问题
如何填充包含复杂数据类型的Cassandra表,例如Pig 0.9.2中的集合或列表
我的具体应用概述:
-
我根据描述创建了相应的Cassandra表:
CREATE TABLE mycassandracf ( my_id int, date timestamp, my_count bigint, grouped_ids list<bigint>, PRIMARY KEY (my_id, date));
-
和STORE指令,携带准备语句:
STORE CassandraAggregate INTO 'cql://test/mycassandracf?output_query=UPDATE+test.mycassandracf+set+my_count+%3D+%3F%2C+grouped_ids+%3D+%3F' USING CqlStorage;
-
根据"GROUP BY"关系,我以cql友好格式(例如元组)"生成"一个关系,并将其存储到Cassandra中。
CassandraAggregate = FOREACH GroupedRelation GENERATE TOTUPLE(TOTUPLE('my_id', $0.my_id), TOTUPLE('date', ISOToUnix($0.createdAt))), TOTUPLE(COUNT($1), $1.grouped_id); DUMP CassandraAggregate; (((my_id,30021),(date,1357084800000)),(2,{(60128490006325819),(62726281032786005)})) (((my_id,30165),(date,1357084800000)),(1,{(60128411174143024)})) (((my_id,30376),(date,1357084800000)),(4,{(60128411146211875),(63645100121476995),(60128411146211875),(63645100121476995)}))
不出所料,在该关系上使用STORE指令会引发异常:
java.lang.ClassCastException:org.apache.pig.data.DefaultDataBag无法强制转换为org.apache.pig.data.DataByteArray
因此,我添加了一个用python编写的UDF,以在grouped_id包上应用一些扁平化:
@outputSchema("flat_bag:bag{}")
def flattenBag(bag):
return tuple([long(item) for tup in bag for item in tup])
我使用tuple是因为使用python集和python列表最终会导致转换错误。
将它添加到我的管道中,我有:
CassandraAggregate = FOREACH GroupedRelation
GENERATE TOTUPLE(TOTUPLE('my_id', $0.my_id),
TOTUPLE('date', ISOToUnix($0.createdAt))),
TOTUPLE(COUNT($1), py_f.flattenBag($1.grouped_id));
DUMP CassandraAggregate;
(((my_id,30021),(date,1357084800000)),(2,(60128490006325819,62726281032786005)))
(((my_id,31120),(date,1357084800000)),(1,(60128411174143024)))
(((my_id,31120),(date,1357084800000)),(1,(60128411146211875,63645100121476995,6012841114621187563645100121476995)))
在最后一个关系上使用STORE指令会引发异常,错误堆栈为:
java.io.IOException: java.io.IOException: org.apache.thrift.transport.TTransportException
at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.runPipeline(PigGenericMapReduce.java:465)
at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.processOnePackageOutput(PigGenericMapReduce.java:428)
at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.reduce(PigGenericMapReduce.java:408)
at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigGenericMapReduce$Reduce.reduce(PigGenericMapReduce.java:262)
at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:176)
at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:652)
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:420)
at org.apache.hadoop.mapred.Child$4.run(Child.java:266)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:415)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1121)
at org.apache.hadoop.mapred.Child.main(Child.java:260)
Caused by: java.io.IOException: org.apache.thrift.transport.TTransportException
at org.apache.cassandra.hadoop.cql3.CqlRecordWriter$RangeClient.run(CqlRecordWriter.java:248)
Caused by: org.apache.thrift.transport.TTransportException
at org.apache.thrift.transport.TIOStreamTransport.read(TIOStreamTransport.java:132)
at org.apache.thrift.transport.TTransport.readAll(TTransport.java:84)
at org.apache.thrift.transport.TFramedTransport.readFrame(TFramedTransport.java:129)
at org.apache.thrift.transport.TFramedTransport.read(TFramedTransport.java:101)
at org.apache.thrift.transport.TTransport.readAll(TTransport.java:84)
at org.apache.thrift.protocol.TBinaryProtocol.readAll(TBinaryProtocol.java:378)
at org.apache.thrift.protocol.TBinaryProtocol.readI32(TBinaryProtocol.java:297)
at org.apache.thrift.protocol.TBinaryProtocol.readMessageBegin(TBinaryProtocol.java:204)
at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:69)
at org.apache.cassandra.thrift.Cassandra$Client.recv_execute_prepared_cql3_query(Cassandra.java:1724)
at org.apache.cassandra.thrift.Cassandra$Client.execute_prepared_cql3_query(Cassandra.java:1709)
at org.apache.cassandra.hadoop.cql3.CqlRecordWriter$RangeClient.run(CqlRecordWriter.java:232)
我用简单的数据类型测试了完全相同的工作流程,并且工作得很完美。我真正想要的是用复杂的类型填充cassandra表的方法,比如Pig中的集合或列表。
非常感谢
经过进一步调查,我在这里找到了解决方案:
https://issues.apache.org/jira/browse/CASSANDRA-5867
基本上,CqlStorage支持复杂的类型。为此,类型应该由元组中的一个元组表示,并将数据类型作为字符串作为第一个元素携带。对于列表,这是如何做到的:
# python
@outputSchema("flat_bag:bag{}")
def flattenBag(bag):
return ('list',) + tuple([long(item) for tup in bag for item in tup])
因此,在咕哝声中:
# pig
CassandraAggregate = FOREACH GroupedRelation
GENERATE TOTUPLE(TOTUPLE('my_id', $0.my_id),
TOTUPLE('date', ISOToUnix($0.createdAt))),
TOTUPLE(COUNT($1), py_f.flattenBag($1.grouped_id));
DUMP CassandraAggregate;
(((my_id,30021),(date,1357084800000)),(2,(list, 60128490006325819,62726281032786005)))
(((my_id,31120),(date,1357084800000)),(1,(list, 60128411174143024)))
(((my_id,31120),(date,1357084800000)),(1,(list, 60128411146211875,63645100121476995,6012841114621187563645100121476995)))
然后使用经典的编码prepared语句将其存储到cassandra中。
希望这会有所帮助。