not resolved
SVBridget
Member
We are using composite columns in the cassandra column family, and those seem to end up mangled on the hive side. We're using dynamic mapping so the composite column name ends up as the value of the 2nd column in the hive table, and it contains basically illegible characters: t�A���à¡>
Does hive not support composite columns in cassandra?
Anonymous
Unregistered
I am also interested in mapping dynamic composite columns to hive. Have you found a way to do this? Is it supported? Will it be supported?
Anonymous
Unregistered
Did you guys figure this out? Can this be done using SerDe via cassandra.cf.validatortype for the CompositeType to something like CompositeType(LongType, UTF8Type)?
tjake
Moderator
This can be done in hive when the next release of DSE comes out. Hive has added a "binary" type so you can map your composite column to binary and write a hive UDF to parse it to do with it what you want.
https://cwiki.apache.org/confluence/display/Hive/HivePlugins
Anonymous
Unregistered
With 2.0, is this fixed? If so, what would be the structure of compositeType to parse it out in Hive UDF?
tjake
Moderator
Yes this is in DSE 2.0, you can set the composite column type as binary
The format of a composite column is explained here: http://www.jointhegrid.com/highperfcassandra/?p=152
Anonymous
Unregistered
I went ahead and created a test example with DSE 2.0. I am still getting an error. Can you please confirm if my mapping is wrong
create keyspace TestData;
-- column family, compositeType columns are (timeinmiilis, string)
create column family TestTable
with key_validation_class = UTF8Type
and comparator = 'CompositeType (LongType, UTF8Type)'
and comment = 'A Test Column Family'
and compression_options={sstable_compression:SnappyCompressor, chunk_length_kb:6};
// test UDF java program for hive, it concatenates the long and string with an under_score in the middle
package com.foo.hive.udf;
import java.nio.ByteBuffer;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.BytesWritable;
public final class TestExtract extends UDF {
public String evaluate(final BytesWritable s) {
if (s == null) {
return null;
}
// byte buffer
ByteBuffer bb = ByteBuffer.wrap(s.getBytes());
// header byte
bb.get();
// first part of composite
int colASize = (bb.get() & 0xFF) << (7 + 1);
colASize = (bb.get() & 0xFF);
byte[] colA = new byte[colASize];
bb.get(colA);
// second part of composite
int colBSize = (bb.get() & 0xFF) << (7 + 1);
colBSize = (bb.get() & 0xFF);
byte[] colB = new byte[colBSize];
bb.get(colB);
// return concatenated string
return new String(new String(colA) + '_' + new String(colB));
}
}
--export the jar
export HIVE_AUX_JARS_PATH=/tmp/custom.jar
--login to hive
use TestData;
-- drop if exists
drop table TestTable;
--create hive mapping
CREATE EXTERNAL TABLE TestTable (row_key string, param_name binary, param_value string)
STORED BY 'org.apache.hadoop.hive.cassandra.CassandraStorageHandler'
WITH SERDEPROPERTIES ("cassandra.ks.name" = "TestData",
"cassandra.columns.mapping" = ":key,:column,:value",
"cassandra.cf.validatortype"= "BytesType, BytesType, BytesType")
TBLPROPERTIES ( "cassandra.ks.name" = "TestData",
"cassandra.range.size" = "100",
"cassandra.slice.predicate.size" = "100");
-- create temp function
create temporary function my_test as 'com.foo.hive.udf.TestExtract';
-- select
select row_key, my_test(param_name), param_value from TestTable limit 5;
--output comes out as
e8cf5f471d0211e18c8e6431504ba7b8 _6 abc123
e8cf5f471d0211e18c8e6431504ba7b8 _6 test132
e8cf5f471d0211e18c8e6431504ba7b8 _6 blah
e8cf5f471d0211e18c8e6431504ba7b8 _6 blah2
e8cf5f471d0211e18c8e6431504ba7b8 _6 hello123
What am I doing wrong here?
Anonymous
Unregistered
Here is what the data looks like from cassandra-cli
(column=1332972763782:Test1, value=abc123, timestamp=1332972760270111)
(column=1332972763782:Test2, value=test132, timestamp=1332972760269241)
(column=1332972763782:Test3, value=blah, timestamp=1332972760270101)
(column=1332972763782:Test4, value=blah2, timestamp=1332972760269340)
(column=1332972763782:Test5, value=hello123, timestamp=1332972760270320)
Anonymous
Unregistered
I got it. Here is the UDF that works
import java.nio.ByteBuffer;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.BytesWritable;
public final class TestExtract extends UDF {
public String evaluate(final BytesWritable s) {
if (s == null) {
return null;
}
// byte buffer
ByteBuffer bb = ByteBuffer.wrap(s.getBytes());
// first part of composite
int colASize = (bb.get() & 0xFF) << (7 + 1);
colASize = (bb.get() & 0xFF);
byte[] colA = new byte[colASize];
bb.get(colA);
ByteBuffer bf = ByteBuffer.wrap(colA);
bb.get();
// second part of composite
int colBSize = (bb.get() & 0xFF) << (7 + 1);
colBSize = (bb.get() & 0xFF);
byte[] colB = new byte[colBSize];
bb.get(colB);
// return concatenated string
return new String(bf.getLong() + "_" + new String(colB));
}
}
dipesh
Member
Thanks alexliu!!
I could get Composite Column working on SELECT statement, however got an exception when tried to write data into Composite Column using following hive script and custom UDF.
Here is relevant log on Exception:
[2013-02-06 22:23:15,252] FATAL {ExecReducer} - org.apache.hadoop.hive.ql.metadata.HiveException: Hive Runtime Error while processing row (tag=0) {"key":{"_col0":"blr_01","_col1":1350465782,"_col2":"WARN"},"value":{"_col0":1},"alias":0}
at org.apache.hadoop.hive.ql.exec.ExecReducer.reduce(ExecReducer.java:256)
at org.apache.hadoop.mapred.ReduceTask.runOldReducer(ReduceTask.java:518)
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:419)
at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:256)
Caused by: org.apache.hadoop.hive.ql.metadata.HiveException: java.io.IOException: InvalidRequestException(why:Not enough bytes to read value of component 0)
at org.apache.hadoop.hive.ql.exec.FileSinkOperator.processOp(FileSinkOperator.java:603)
...
Here is my hive script:
CREATE TEMPORARY FUNCTION setCompositeLongString AS
'com.cisco.iep.hive.plugins.WriteCompositeLongString';
INSERT OVERWRITE TABLE TestTable
SELECT src_id,
setCompositeLongString(cast(alert_time as BIGINT), severity),
COUNT(alert_type)
FROM alert_instance
GROUP BY src_id,
cast(alert_time as BIGINT),
severity;And here is custom UDF to write data into Composite Column:
package com.cisco.iep.hive.plugins;
import java.io.ByteArrayOutputStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
public class WriteCompositeLongString extends UDF{
public BytesWritable evaluate(final LongWritable l, final Text s) {
List<byte[]> b = new ArrayList<byte[]>(2);
b.add(longWritableToByteArray(l));
b.add(textToByteArray(s));
return new BytesWritable(makeComposite(b));
}
private byte[] textToByteArray(Text s){
if(s==null){
return null;
}
return s.getBytes();
}
private byte[] longWritableToByteArray(final LongWritable l){
if(l==null){
return null;
}
return ByteBuffer.allocate(Long.SIZE).putLong(l.get()).array();
}
private byte[] makeComposite(final List<byte[]> b) {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
for (int i = 0; i < b.size(); i++) {
bos.write((byte) ((b.get(i).length >> (7 + 1)) & 0xFF));
bos.write((byte) (b.get(i).length & 0xFF));
for (int j = 0; j < b.get(i).length; j++) {
bos.write(b.get(i)[j] & 0xFF);
}
bos.write((byte) 0);
}
return bos.toByteArray();
}
}Please advise.
alexliu
Member
Here is composite column encoding according to the source code
/*
* The encoding of a CompositeType column name should be:
* <component><component><component> ...
* where <component> is:
* <length of value><value><'end-of-component' byte>
* where <length of value> is a 2 bytes unsigned short the and the
* 'end-of-component' byte should always be 0 for actual column name.
* However, it can set to 1 for query bounds. This allows to query for the
* equivalent of 'give me the full super-column'. That is, if during a slice
* query uses:
* start = <3><"foo".getBytes()><0>
* end = <3><"foo".getBytes()><1>
* then he will be sure to get *all* the columns whose first component is "foo".
* If for a component, the 'end-of-component' is != 0, there should not be any
* following component. The end-of-component can also be -1 to allow
* non-inclusive query. For instance:
* start = <3><"foo".getBytes()><-1>
* allows to query everything that is greater than <3><"foo".getBytes()>, but
* not <3><"foo".getBytes()> itself.
*/
here is a sample code to create encoding composite column
private ByteBuffer createCompositeKey(String s, UUID uuid, int i, boolean lastIsOne)
{
ByteBuffer bytes = ByteBufferUtil.bytes(s);
int totalSize = 0;
if (s != null)
{
totalSize += 2 + bytes.remaining() + 1;
if (uuid != null)
{
totalSize += 2 + 16 + 1;
if (i != -1)
{
totalSize += 2 + 1 + 1;
}
}
}
ByteBuffer bb = ByteBuffer.allocate(totalSize);
if (s != null)
{
bb.putShort((short) bytes.remaining());
bb.put(bytes);
bb.put(uuid == null && lastIsOne ? (byte)1 : (byte)0);
if (uuid != null)
{
bb.putShort((short) 16);
bb.put(UUIDGen.decompose(uuid));
bb.put(i == -1 && lastIsOne ? (byte)1 : (byte)0);
if (i != -1)
{
// We are putting a byte only because our test use ints that fit in a byte *and* IntegerType.fromString() will
// return something compatible (i.e, putting a full int here would break 'fromStringTest')
bb.putShort((short) 1);
bb.put((byte)i);
bb.put(lastIsOne ? (byte)1 : (byte)0);
}
}
}
bb.rewind();
return bb;
}
You need change your code to the same encoding as the composite column name
You must log in to post.