Mailing-List: contact user-help@cassandra.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@cassandra.apache.org
Received-SPF: pass (nike.apache.org: domain of penela@gmail.com designates
 74.125.82.44 as permitted sender)
MIME-Version: 1.0
Date: Mon, 27 Aug 2012 13:09:57 +0200
Message-ID: 
 <CAMbHBZ9NNiiK2x49EhSCw-EfBPJdY03gy7idsjPPuQLErZcR6g@mail.gmail.com>
Subject: Understanding Cassandra + MapReduce + Composite Columns
From: =?ISO-8859-1?Q?V=EDctor_Penela?= <penela@gmail.com>
To: user@cassandra.apache.org
Content-Type: multipart/alternative; boundary=f46d043c06e6aef59904c83d5d74

--f46d043c06e6aef59904c83d5d74
Content-Type: text/plain; charset=ISO-8859-1

Hi!

I'm trying to use Hadoop's MapReduce on top of a Cassandra environment, and
I've running into some issues while using Composite Columns. I'm currently
using Cassandra 1.1.2 (I wouldn't mind having to update it) and Hadoop
1.0.3 (I'd rather keep this version).

What I would like to do is send slices divided by the first key of the
composite key, and do some processing, taking into account the rest of the
elements of the composite key (as well as other columns).

I've built a sandbox keyspace with some column families in order to test
this:

CREATE TABLE test_1 (
  field1 text,
  field2 text,
  field3 text,
  field4 text,
  PRIMARY KEY (field1)
) ;
CREATE TABLE test_2 (
  field1 text,
  field2 text,
  field3 text,
  field4 text,
  PRIMARY KEY (field1, field2)
) ;

The Job configuration (the relevant elements for Cassandra) is as follows:
// Cassandra config
ConfigHelper.setInputRpcPort(conf, "9160");
ConfigHelper.setInputInitialAddress(conf, "localhost");
ConfigHelper.setInputPartitioner(conf, "ByteOrderedPartitioner");
ConfigHelper.setInputColumnFamily(conf, KEYSPACE, INPUT_COLUMN_FAMILY);

SlicePredicate predicate = new SlicePredicate();
predicate.setSlice_range(new
SliceRange().setStart(ByteBufferUtil.EMPTY_BYTE_BUFFER).setFinish(ByteBufferUtil.EMPTY_BYTE_BUFFER).setCount(5));
ConfigHelper.setInputSlicePredicate(conf, predicate);

My dummy maps tries only to log the different keys and values received.
map(ByteBuffer key, SortedMap<ByteBuffer, IColumn> columns,
OutputCollector<Text, IntWritable> output,  Reporter reporter) { ... }

With CF test_1 everything seems to work fine.

With CF test_2, I only receive field1 value inside the ByteBuffer key. The
rest of the composite key seems to be encoded into each key of the
SortedMap with the particular key of that column (field3, field4, ...), but
I don't know exactly how to extract it (I'm a bit new with ByteBuffers, so
any help there will be welcome :)). Is there anyway to specify the schema
of this particular CF at MR level, in order to be able to extract the
secondary key?

Thanks!

--f46d043c06e6aef59904c83d5d74
Content-Type: text/html; charset=ISO-8859-1
Content-Transfer-Encoding: quoted-printable

Hi!<div><br></div><div>I&#39;m trying to use Hadoop&#39;s MapReduce on top =
of a Cassandra environment, and I&#39;ve running into some issues while usi=
ng Composite Columns. I&#39;m currently using Cassandra 1.1.2 (I wouldn&#39=
;t mind having to update it) and Hadoop 1.0.3 (I&#39;d rather keep this ver=
sion).</div>

<div><br></div><div>What I would like to do is send slices divided by the f=
irst key of the composite key, and do some processing, taking into account =
the rest of the elements of the composite key (as well as other columns).</=
div>

<div><br></div><div>I&#39;ve built a sandbox keyspace with some column fami=
lies in order to test this:</div><div><br></div><div><div>CREATE TABLE test=
_1 (</div><div>=A0 field1 text,</div><div>=A0 field2 text,</div><div>=A0 fi=
eld3 text,</div>

<div>=A0 field4 text,</div><div>=A0 PRIMARY KEY (field1)</div><div>) ;</div=
><div>CREATE TABLE test_2 (</div><div>=A0 field1 text,</div><div>=A0 field2=
 text,</div><div>=A0 field3 text,</div><div>=A0 field4 text,</div><div>=A0 =
PRIMARY KEY (field1, field2)</div>

<div>) ;</div></div><div><br></div><div>The Job configuration (the relevant=
 elements for Cassandra) is as follows:</div><div><div>// Cassandra config<=
/div><div>ConfigHelper.setInputRpcPort(conf, &quot;9160&quot;);</div><div>

ConfigHelper.setInputInitialAddress(conf, &quot;localhost&quot;);</div><div=
>ConfigHelper.setInputPartitioner(conf, &quot;ByteOrderedPartitioner&quot;)=
;</div><div>ConfigHelper.setInputColumnFamily(conf, KEYSPACE, INPUT_COLUMN_=
FAMILY);</div>

<div><br></div><div>SlicePredicate predicate =3D new SlicePredicate();</div=
><div>predicate.setSlice_range(new SliceRange().setStart(ByteBufferUtil.EMP=
TY_BYTE_BUFFER).setFinish(ByteBufferUtil.EMPTY_BYTE_BUFFER).setCount(5));</=
div>

<div>ConfigHelper.setInputSlicePredicate(conf, predicate);</div></div><div>=
<br></div><div>My dummy maps tries only to log the different keys and value=
s received.</div><div>map(ByteBuffer key, SortedMap&lt;ByteBuffer, IColumn&=
gt; columns, OutputCollector&lt;Text, IntWritable&gt; output, =A0Reporter r=
eporter) { ... }</div>

<div><br></div><div>With CF test_1 everything seems to work fine.</div><div=
><br></div><div>With CF test_2, I only receive field1 value inside the Byte=
Buffer key. The rest of the composite key seems to be encoded into each key=
 of the SortedMap with the particular key of that column (field3, field4, .=
..), but I don&#39;t know exactly how to extract it (I&#39;m a bit new with=
 ByteBuffers, so any help there will be welcome :)). Is there anyway to spe=
cify the schema of this particular CF at MR level, in order to be able to e=
xtract the secondary key?</div>

<div><br></div><div>Thanks!</div>

--f46d043c06e6aef59904c83d5d74--