From user-return-5466-apmail-cassandra-user-archive=cassandra.apache.org@cassandra.apache.org Thu May 06 23:36:33 2010 Return-Path: Delivered-To: apmail-cassandra-user-archive@www.apache.org Received: (qmail 87686 invoked from network); 6 May 2010 23:36:33 -0000 Received: from unknown (HELO mail.apache.org) (140.211.11.3) by 140.211.11.9 with SMTP; 6 May 2010 23:36:33 -0000 Received: (qmail 45136 invoked by uid 500); 6 May 2010 23:36:32 -0000 Delivered-To: apmail-cassandra-user-archive@cassandra.apache.org Received: (qmail 45120 invoked by uid 500); 6 May 2010 23:36:32 -0000 Mailing-List: contact user-help@cassandra.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: user@cassandra.apache.org Delivered-To: mailing list user@cassandra.apache.org Received: (qmail 45112 invoked by uid 99); 6 May 2010 23:36:32 -0000 Received: from athena.apache.org (HELO athena.apache.org) (140.211.11.136) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 06 May 2010 23:36:32 +0000 X-ASF-Spam-Status: No, hits=2.2 required=10.0 tests=HTML_MESSAGE,RCVD_IN_DNSWL_NONE X-Spam-Check-By: apache.org Received-SPF: unknown (athena.apache.org: error in processing during lookup of mark.schnitzius@cxense.com) Received: from [209.85.223.201] (HELO mail-iw0-f201.google.com) (209.85.223.201) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 06 May 2010 23:36:27 +0000 Received: by iwn39 with SMTP id 39so650661iwn.7 for ; Thu, 06 May 2010 16:36:04 -0700 (PDT) MIME-Version: 1.0 Received: by 10.231.191.73 with SMTP id dl9mr1096074ibb.50.1273188963254; Thu, 06 May 2010 16:36:03 -0700 (PDT) Received: by 10.231.183.135 with HTTP; Thu, 6 May 2010 16:36:03 -0700 (PDT) In-Reply-To: <1273184762.184514166@192.168.2.230> References: <1273184762.184514166@192.168.2.230> Date: Fri, 7 May 2010 09:36:03 +1000 Message-ID: Subject: Re: Updating (as opposed to just setting) Cassandra data via Hadoop From: Mark Schnitzius To: user@cassandra.apache.org Content-Type: multipart/alternative; boundary=0016367d4eace303970485f567bd --0016367d4eace303970485f567bd Content-Type: text/plain; charset=ISO-8859-1 Yes, I think this approach would be more efficient, but Ian's point about failed runs is well taken. It is still a problem with this approach. I may have to introduce a scheme where Hadoop's output is written to a new column family, and then some sort of pointer is updated to point to this column family at the end of a successful run. That way, if a run failed, the pointer would still point to the old data, and we could just rerun the update task. (Older and failed versions of the data would have to be periodically purged as well.) The other difficulty with this approach is massaging the existing Cassandra data (which will come in as >) to look like whatever the other input stream looks like, since a Hadoop mapper can only take one key/value type. Some sort of wrapper class should work, but the input streams might even have different granularities (i.e. each entry in the SortedMap> map might correspond to a single input row from the other input stream). I'll put something on the wiki if I can make it work... Cheers Mark On Fri, May 7, 2010 at 8:26 AM, Stu Hood wrote: > Ian: I think that as get_range_slice gets faster, the approach that Mark > was heading toward may be considerably more efficient than reading the old > value in the OutputFormat. > > Mark: Reading all of the data you want to update out of Cassandra using the > InputFormat, merging it with (tagged) new data, and then performing > deletes/inserts (without reads) in a Cassandra OutputFormat makes a lot of > sense to me. It gives you much better data locality: all reads happen in a > streaming fashion (mostly: this is where we need improvement for > get_range_slices) at the beginning of the job. > > > -----Original Message----- > From: "Ian Kallen" > Sent: Thursday, May 6, 2010 5:14pm > To: user@cassandra.apache.org > Subject: Re: Updating (as opposed to just setting) Cassandra data via > Hadoop > > I have inputs that are text logs and I wrote a Cassandra OutputFormat, the > reducers read the old values from their respective column families, > increment the counts and write back the new values. Since all of the writes > are done by the hadoop jobs and we're not running multiple jobs > concurrently, the writes aren't clobbering any values except the ones from > the prior hadoop run. If/when atomic increments are available, we'd be able > to run concurrent log processing jobs for but for now, this seems to work. > I > think the biggest risk is that a reduce task fails, hadoop restarts it and > the replacement task re-increments the values. We're going to wait and see > to determine how much of a problem this is in practice. > -Ian > > On Tue, May 4, 2010 at 8:53 PM, Mark Schnitzius > wrote: > > > I have a situation where I need to accumulate values in Cassandra on an > > ongoing basis. Atomic increments are still in the works apparently (see > > https://issues.apache.org/jira/browse/CASSANDRA-721) so for the time > being > > I'll be using Hadoop, and attempting to feed in both the existing values > and > > the new values to a M/R process where they can be combined together and > > written back out to Cassandra. > > > > The approach I'm taking is to use Hadoop's CombineFileInputFormat to > blend > > the existing data (using Cassandra's ColumnFamilyInputFormat) with the > newly > > incoming data (using something like Hadoop's SequenceFileInputFormat). > > > > I was just wondering, has anyone here tried this, and were there issues? > > I'm worried because the CombineFileInputFormat has restrictions around > > splits being from different pools so I don't know how this will play out > > with data from both Cassandra and HDFS. The other option, I suppose, is > to > > use a separate M/R process to replicate the data onto HDFS first, but I'd > > rather avoid the extra step and duplication of storage. > > > > Also, if you've tackled a similar situation in the past using a different > > approach, I'd be keen to hear about it... > > > > > > Thanks > > Mark > > > > > --0016367d4eace303970485f567bd Content-Type: text/html; charset=ISO-8859-1 Content-Transfer-Encoding: quoted-printable Yes, I think this approach would be more efficient, but Ian's point abo= ut failed runs is well taken. =A0It is still a problem with this approach. = =A0I may have to introduce a scheme where Hadoop's output is written to= a new column family, and then some sort of pointer is updated to point to = this column family at the end of a successful run. =A0That way, if a run fa= iled, the pointer would still point to the old data, and we could just reru= n the update task. =A0(Older and failed versions of the data would have to = be periodically purged as well.)

The other difficulty with this approach is massaging the exi= sting Cassandra data (which will come in as <String, SortedMap<byte[]= , IColumn>>) to look like whatever the other input stream looks like,= since a Hadoop mapper can only take one key/value type. =A0Some sort of wr= apper class should work, but the input streams might even have different gr= anularities (i.e. each entry in the=A0SortedMap<byte[], IColumn>> = map might correspond to a single input row from the other input stream).

I'll put something on the wiki if I can make it wor= k...


Cheers
Mark
=

On Fri, May 7, 2010 at 8:26 AM, Stu Hood <stu.hood@racks= pace.com> wrote:
Ian: I think that as get_range_slice gets faster, the approach that Mark wa= s heading toward may be considerably more efficient than reading the old va= lue in the OutputFormat.

Mark: Reading all of the data you want to update out of Cassandra using the= InputFormat, merging it with (tagged) new data, and then performing delete= s/inserts (without reads) in a Cassandra OutputFormat makes a lot of sense = to me. It gives you much better data locality: all reads happen in a stream= ing fashion (mostly: this is where we need improvement for get_range_slices= ) at the beginning of the job.


-----Original Message-----
From: "Ian Kallen" <spidaman.list@gmail.com>
Sent: Thursday, May 6, 2010 5:14pm
To: user@cassandra.apache.org<= /a>
Subject: Re: Updating (as opposed to just setting) Cassandra data via Hadoo= p

I have inputs that are text logs and I wrote a Cassandra OutputFormat, the<= br> reducers read the old values from their respective column families,
increment the counts and write back the new values. Since all of the writes=
are done by the hadoop jobs and we're not running multiple jobs
concurrently, the writes aren't clobbering any values except the ones f= rom
the prior hadoop run. If/when atomic increments are available, we'd be = able
to run concurrent log processing jobs for but for now, this seems to work. = I
think the biggest risk is that a reduce task fails, hadoop restarts it and<= br> the replacement task re-increments the values. We're going to wait and = see
to determine how much of a problem this is in practice.
-Ian

On Tue, May 4, 2010 at 8:53 PM, Mark Schnitzius
<mark.schnitzius@cxense.co= m>wrote:

> I have a situation where I need to accumulate values in Cassandra on a= n
> ongoing basis. =A0Atomic increments are still in the works apparently = (see
> https://issues.apache.org/jira/browse/CASSANDRA-721) so for= the time being
> I'll be using Hadoop, and attempting to feed in both the existing = values and
> the new values to a M/R process where they can be combined together an= d
> written back out to Cassandra.
>
> The approach I'm taking is to use Hadoop's CombineFileInputFor= mat to blend
> the existing data (using Cassandra's ColumnFamilyInputFormat) with= the newly
> incoming data (using something like Hadoop's SequenceFileInputForm= at).
>
> I was just wondering, has anyone here tried this, and were there issue= s?
> =A0I'm worried because the CombineFileInputFormat has restrictions= around
> splits being from different pools so I don't know how this will pl= ay out
> with data from both Cassandra and HDFS. =A0The other option, I suppose= , is to
> use a separate M/R process to replicate the data onto HDFS first, but = I'd
> rather avoid the extra step and duplication of storage.
>
> Also, if you've tackled a similar situation in the past using a di= fferent
> approach, I'd be keen to hear about it...
>
>
> Thanks
> Mark
>



--0016367d4eace303970485f567bd--