Hi,
I need to join two HBase tables. The obvious way is to use a M/R job for
that. The problem is that the few references to that question I found
recommend pulling one table to the mapper and then do a lookup for the
referred row in the second table.
This sounds like a very inefficient way to do  join with map reduce. I
believe it would be much better to feed the rows of both tables to the
mapper and let it emit a key based on the join fields. Since all the rows
with the same join fields values will have the same key the reducer will be
able to easily generate the result of the join.
The problem with this is that I couldn't find a way to feed two tables to a
single map reduce job. I could probably dump the tables to files in a single
directory and then run the join on the files but that really makes no sense.

Am I missing something? Any other ideas?

-eran

As far as I can tell there is not yet a build-in mechanism you can use
for this. You could implement your own InputFormat, something like
MultiTableInputFormat. If you need different map functions for the two
tables, perhaps something similar to Hadoop's MultipleInputs should do
the trick.

On 05/31/2011 02:06 PM, Eran Kutner wrote:
> Hi,
> I need to join two HBase tables. The obvious way is to use a M/R job for
> that. The problem is that the few references to that question I found
> recommend pulling one table to the mapper and then do a lookup for the
> referred row in the second table.
> This sounds like a very inefficient way to do  join with map reduce. I
> believe it would be much better to feed the rows of both tables to the
> mapper and let it emit a key based on the join fields. Since all the rows
> with the same join fields values will have the same key the reducer will be
> able to easily generate the result of the join.
> The problem with this is that I couldn't find a way to feed two tables to a
> single map reduce job. I could probably dump the tables to files in a single
> directory and then run the join on the files but that really makes no sense.
>
> Am I missing something? Any other ideas?
>
> -eran
>

MutipleInputs would be ideal, but that seems pretty complicated.
MultiTableInputFormat seems like a simple change in the getSplits() method
of TableInputFormat + support for a collection of table and their matching
scanners instead of a single table and scanner, doesn't sound too
complicated.
Any other suggestions?

-eran

On Tue, May 31, 2011 at 15:31, Ferdy Galema <[EMAIL PROTECTED]>wrote:

> As far as I can tell there is not yet a build-in mechanism you can use for
> this. You could implement your own InputFormat, something like
> MultiTableInputFormat. If you need different map functions for the two
> tables, perhaps something similar to Hadoop's MultipleInputs should do the
> trick.
>
>
> On 05/31/2011 02:06 PM, Eran Kutner wrote:
>
>> Hi,
>> I need to join two HBase tables. The obvious way is to use a M/R job for
>> that. The problem is that the few references to that question I found
>> recommend pulling one table to the mapper and then do a lookup for the
>> referred row in the second table.
>> This sounds like a very inefficient way to do  join with map reduce. I
>> believe it would be much better to feed the rows of both tables to the
>> mapper and let it emit a key based on the join fields. Since all the rows
>> with the same join fields values will have the same key the reducer will
>> be
>> able to easily generate the result of the join.
>> The problem with this is that I couldn't find a way to feed two tables to
>> a
>> single map reduce job. I could probably dump the tables to files in a
>> single
>> directory and then run the join on the files but that really makes no
>> sense.
>>
>> Am I missing something? Any other ideas?
>>
>> -eran
>>
>>

Eran,

You want to join two tables? The short answer is to use a relational database to solve that problem.

Longer answer:

You're using HBase so you don't need to think in terms of a reducer.
You can create a temp table for your query.
You can then run one map job to scan and filter table A, dumping the result set in to the temp table
In parallel, you run a map job to scan and filter table B, dumping the result set in to the temp table.

Voila! You're done. Just remember to clean up and drop the temp table when you're done.

But there may be a problem.
If you use the same column name but the data means different things.  Like both tables have a column named 'Tim' (and why you would name something Tim is beyond me... ;-) ) but this column means one thing in table A and something else in table B and you want to retain both values... You just need to create a column whose name is based on ${tablename}+'|'+${column name} so it would be TableA|Tim and TableB|Tim.

HTH

-Mike
> From: [EMAIL PROTECTED]
> Date: Tue, 31 May 2011 15:43:43 +0300
> Subject: Re: How to efficiently join HBase tables?
> To: [EMAIL PROTECTED]
> CC: [EMAIL PROTECTED]
>
> MutipleInputs would be ideal, but that seems pretty complicated.
> MultiTableInputFormat seems like a simple change in the getSplits() method
> of TableInputFormat + support for a collection of table and their matching
> scanners instead of a single table and scanner, doesn't sound too
> complicated.
> Any other suggestions?
>
> -eran
>
>
>
> On Tue, May 31, 2011 at 15:31, Ferdy Galema <[EMAIL PROTECTED]>wrote:
>
> > As far as I can tell there is not yet a build-in mechanism you can use for
> > this. You could implement your own InputFormat, something like
> > MultiTableInputFormat. If you need different map functions for the two
> > tables, perhaps something similar to Hadoop's MultipleInputs should do the
> > trick.
> >
> >
> > On 05/31/2011 02:06 PM, Eran Kutner wrote:
> >
> >> Hi,
> >> I need to join two HBase tables. The obvious way is to use a M/R job for
> >> that. The problem is that the few references to that question I found
> >> recommend pulling one table to the mapper and then do a lookup for the
> >> referred row in the second table.
> >> This sounds like a very inefficient way to do  join with map reduce. I
> >> believe it would be much better to feed the rows of both tables to the
> >> mapper and let it emit a key based on the join fields. Since all the rows
> >> with the same join fields values will have the same key the reducer will
> >> be
> >> able to easily generate the result of the join.
> >> The problem with this is that I couldn't find a way to feed two tables to
> >> a
> >> single map reduce job. I could probably dump the tables to files in a
> >> single
> >> directory and then run the join on the files but that really makes no
> >> sense.
> >>
> >> Am I missing something? Any other ideas?
> >>
> >> -eran
> >>
> >>
     

Re:  "The problem is that the few references to that question I found recommend pulling one table to the mapper and then do a lookup for the referred row in the second table."

With multi-get in .90.x you could perform some reasonably clever processing and not do the lookups one-by-one but in batches.

Also, if the other table is "small" you could have the leverage the block cache on the lookups (i.e., if it's a domain/lookup table).  

-----Original Message-----
From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran Kutner
Sent: Tuesday, May 31, 2011 8:06 AM
To: [EMAIL PROTECTED]
Subject: How to efficiently join HBase tables?

Hi,
I need to join two HBase tables. The obvious way is to use a M/R job for that. The problem is that the few references to that question I found recommend pulling one table to the mapper and then do a lookup for the referred row in the second table.
This sounds like a very inefficient way to do  join with map reduce. I believe it would be much better to feed the rows of both tables to the mapper and let it emit a key based on the join fields. Since all the rows with the same join fields values will have the same key the reducer will be able to easily generate the result of the join.
The problem with this is that I couldn't find a way to feed two tables to a single map reduce job. I could probably dump the tables to files in a single directory and then run the join on the files but that really makes no sense.

Am I missing something? Any other ideas?

-eran

Maybe I'm missing something... but this isn't a hard problem to solve.

Eran wants to join two tables.
If we look at an SQL Statement...

SELECT A.*, B.*
FROM A, B
WHERE A.1 = B.1
AND  A.2 = B.2
AND  A.3 = xxx
AND A.4 = yyy
AND B.45 = zzz

Or something along those lines.

So what you're essentially doing is saying I want to take a subset of data from table A, and a subset of data from table B and join them on the values in columns 1 and 2.
Table A's data will be filtered on columns 3 and 4 and B's data will be filtered on column 45. NOTE: since you don't know the relationship of the column names to either table, you're safer in writing tableA|column_name and tableB|column_name to your temp table.

So if you create a temp table FOO where the key is column 1 and column 2 (column1|column2) then when you walk through the subsets adding them to the temp table, you
will get the end result automatically.

Then you can output your hbase temp table and then truncate the table.

So what am I missing?

-Mike
> From: [EMAIL PROTECTED]
> To: [EMAIL PROTECTED]
> Date: Tue, 31 May 2011 10:22:34 -0400
> Subject: RE: How to efficiently join HBase tables?
>
>
> Re:  "The problem is that the few references to that question I found recommend pulling one table to the mapper and then do a lookup for the referred row in the second table."
>
> With multi-get in .90.x you could perform some reasonably clever processing and not do the lookups one-by-one but in batches.
>
> Also, if the other table is "small" you could have the leverage the block cache on the lookups (i.e., if it's a domain/lookup table).  
>
>
>
> -----Original Message-----
> From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran Kutner
> Sent: Tuesday, May 31, 2011 8:06 AM
> To: [EMAIL PROTECTED]
> Subject: How to efficiently join HBase tables?
>
> Hi,
> I need to join two HBase tables. The obvious way is to use a M/R job for that. The problem is that the few references to that question I found recommend pulling one table to the mapper and then do a lookup for the referred row in the second table.
> This sounds like a very inefficient way to do  join with map reduce. I believe it would be much better to feed the rows of both tables to the mapper and let it emit a key based on the join fields. Since all the rows with the same join fields values will have the same key the reducer will be able to easily generate the result of the join.
> The problem with this is that I couldn't find a way to feed two tables to a single map reduce job. I could probably dump the tables to files in a single directory and then run the join on the files but that really makes no sense.
>
> Am I missing something? Any other ideas?
>
> -eran
     

Eran's observation was that a join is solvable in a Mapper via lookups on a 2nd HBase table, but it might not be that efficient if the lookups are 1 by 1.  I agree with that.

My suggestion was to use multi-Get for the lookups instead.  So you'd hold onto a batch of records in the Mapper and then the batch size is filled, then you do the lookups (and then any required emitting, etc.).  

-----Original Message-----
From: Michael Segel [mailto:[EMAIL PROTECTED]]
Sent: Tuesday, May 31, 2011 10:56 AM
To: [EMAIL PROTECTED]
Subject: RE: How to efficiently join HBase tables?
Maybe I'm missing something... but this isn't a hard problem to solve.

Eran wants to join two tables.
If we look at an SQL Statement...

SELECT A.*, B.*
FROM A, B
WHERE A.1 = B.1
AND  A.2 = B.2
AND  A.3 = xxx
AND A.4 = yyy
AND B.45 = zzz

Or something along those lines.

So what you're essentially doing is saying I want to take a subset of data from table A, and a subset of data from table B and join them on the values in columns 1 and 2.
Table A's data will be filtered on columns 3 and 4 and B's data will be filtered on column 45. NOTE: since you don't know the relationship of the column names to either table, you're safer in writing tableA|column_name and tableB|column_name to your temp table.

So if you create a temp table FOO where the key is column 1 and column 2 (column1|column2) then when you walk through the subsets adding them to the temp table, you will get the end result automatically.

Then you can output your hbase temp table and then truncate the table.

So what am I missing?

-Mike
> From: [EMAIL PROTECTED]
> To: [EMAIL PROTECTED]
> Date: Tue, 31 May 2011 10:22:34 -0400
> Subject: RE: How to efficiently join HBase tables?
>
>
> Re:  "The problem is that the few references to that question I found recommend pulling one table to the mapper and then do a lookup for the referred row in the second table."
>
> With multi-get in .90.x you could perform some reasonably clever processing and not do the lookups one-by-one but in batches.
>
> Also, if the other table is "small" you could have the leverage the block cache on the lookups (i.e., if it's a domain/lookup table).  
>
>
>
> -----Original Message-----
> From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran
> Kutner
> Sent: Tuesday, May 31, 2011 8:06 AM
> To: [EMAIL PROTECTED]
> Subject: How to efficiently join HBase tables?
>
> Hi,
> I need to join two HBase tables. The obvious way is to use a M/R job for that. The problem is that the few references to that question I found recommend pulling one table to the mapper and then do a lookup for the referred row in the second table.
> This sounds like a very inefficient way to do  join with map reduce. I believe it would be much better to feed the rows of both tables to the mapper and let it emit a key based on the join fields. Since all the rows with the same join fields values will have the same key the reducer will be able to easily generate the result of the join.
> The problem with this is that I couldn't find a way to feed two tables to a single map reduce job. I could probably dump the tables to files in a single directory and then run the join on the files but that really makes no sense.
>
> Am I missing something? Any other ideas?
>
> -eran
     

Thanks everyone for the great feedback. I'll try to address all the
suggestions.

My data sets go between large and very large. One is in the order of many
billions of rows, although the input for a typical MR job will be in the
hundreds of millions, the second table is in the tens of millions. I doubt a
SQL DB will handle this kind of a join in a reasonable manner.

Doing batched lookup will indeed be more efficient than one by one but it
will require the mapper to manage a local state between multiple calls,
which is something I don't really like doing, and worse, it doesn't really
solve the lookup problem it only moves it one tier lower. Instead of the
mapper having to do all thouse random lookups, now HBase itself will have to
do them. Granted it is more efficient than individual lookup API calls but
it is not nearly as efficient as doing sequential reads.

Finally, the temp table method, that will work but again, I suspect it will
be a lot less efficient than the sequence files Hadoop would generate. The
join output is expected to be in the tens of millions of rows, each with
multiple columns. From some tests I've done, writing this number of rows to
a clean table starts out very slowly and takes a lot of time to ramp up as
the regions begin to split and move around the cluster. I should say that
the output of this join is just the input for another MR job, so it would
really be just a temp table and not something that would be useful after
that.

I should also say that I have looked into eliminating the lookup altogether
by resolving the data from the second table before the rows are inserted to
the main table, kind of denormalization, but that would introduce an
unacceptable latency to a very high volume process.

Still looking for other ideas.

-eran

On Tue, May 31, 2011 at 18:42, Doug Meil <[EMAIL PROTECTED]>wrote:

> Eran's observation was that a join is solvable in a Mapper via lookups on a
> 2nd HBase table, but it might not be that efficient if the lookups are 1 by
> 1.  I agree with that.
>
> My suggestion was to use multi-Get for the lookups instead.  So you'd hold
> onto a batch of records in the Mapper and then the batch size is filled,
> then you do the lookups (and then any required emitting, etc.).
>
>
>
> -----Original Message-----
> From: Michael Segel [mailto:[EMAIL PROTECTED]]
> Sent: Tuesday, May 31, 2011 10:56 AM
> To: [EMAIL PROTECTED]
> Subject: RE: How to efficiently join HBase tables?
>
>
> Maybe I'm missing something... but this isn't a hard problem to solve.
>
> Eran wants to join two tables.
> If we look at an SQL Statement...
>
> SELECT A.*, B.*
> FROM A, B
> WHERE A.1 = B.1
> AND  A.2 = B.2
> AND  A.3 = xxx
> AND A.4 = yyy
> AND B.45 = zzz
>
> Or something along those lines.
>
> So what you're essentially doing is saying I want to take a subset of data
> from table A, and a subset of data from table B and join them on the values
> in columns 1 and 2.
> Table A's data will be filtered on columns 3 and 4 and B's data will be
> filtered on column 45. NOTE: since you don't know the relationship of the
> column names to either table, you're safer in writing tableA|column_name and
> tableB|column_name to your temp table.
>
> So if you create a temp table FOO where the key is column 1 and column 2
> (column1|column2) then when you walk through the subsets adding them to the
> temp table, you will get the end result automatically.
>
> Then you can output your hbase temp table and then truncate the table.
>
> So what am I missing?
>
> -Mike
>
>
> > From: [EMAIL PROTECTED]
> > To: [EMAIL PROTECTED]
> > Date: Tue, 31 May 2011 10:22:34 -0400
> > Subject: RE: How to efficiently join HBase tables?
> >
> >
> > Re:  "The problem is that the few references to that question I found
> recommend pulling one table to the mapper and then do a lookup for the
> referred row in the second table."
> >
> > With multi-get in .90.x you could perform some reasonably clever
> processing and not do the lookups one-by-one but in batches.

Doesn't Hive for HBase enable joins?

On Tue, May 31, 2011 at 5:06 AM, Eran Kutner <[EMAIL PROTECTED]> wrote:
> Hi,
> I need to join two HBase tables. The obvious way is to use a M/R job for
> that. The problem is that the few references to that question I found
> recommend pulling one table to the mapper and then do a lookup for the
> referred row in the second table.
> This sounds like a very inefficient way to do  join with map reduce. I
> believe it would be much better to feed the rows of both tables to the
> mapper and let it emit a key based on the join fields. Since all the rows
> with the same join fields values will have the same key the reducer will be
> able to easily generate the result of the join.
> The problem with this is that I couldn't find a way to feed two tables to a
> single map reduce job. I could probably dump the tables to files in a single
> directory and then run the join on the files but that really makes no sense.
>
> Am I missing something? Any other ideas?
>
> -eran
>

Doug,

I read the OP's post as the following:
"> Hi,
> I need to join two HBase tables.
The obvious way is to use a M/R job for that.
The problem is that the few references to that question I found recommend pulling one table to the mapper
and then do a lookup for the referred row in the second table.
"

So the underlying problem that the OP was trying to solve was how to join two tables from HBase.
Unfortunately I goofed.
I gave a quick and dirty solution that is a bit incomplete. They row key in the temp table has to be unique and I forgot about the Cartesian
product. So my solution wouldn't work in the general case.

If you wanted to do this in the general case...

Map job 1 selects data from the first table and then writes it to tempA where the key is
the join column(s) in order, followed by all of the rows. (This is a wide key)
Also the map job returns a count of all of the rows in tempA

Map job 2 selects data from the second table again writing the entire return set as the key so that each row is also unique.

Now take the smaller of two sets and run a map reduce job.
For each row, you'd want to create a scan of the other table, using a start and stop row based on the value in the joined columns.
Write out that row and the corresponding row from the scan in to your result set... but here I think you just need a unique key if you're just
dumping this to file, otherwise you will want to write the entire record as your key and each column as your value.

This will account for the Cartesian product in the join.

So it will look something like this:
Tab A
K, 1,2,3
   a,b,c
   a,a,b
   b,b,a
   b,c,a
   c,b,a

Tab B
K, 1,2,3
   a,e,f
   a,c,d
   b,z,z

SELECT *
FROM A, B
WHERE A.1 = B.1
AND   A.1 = a
Tab A
K 1,2,3
a|b|c a,b,c
a|a|b a,a,b

Tab B
a|e|f a,e,f
a|c|d a,c,d

So output =
K 1,2,3,4,5,6  
a|b|c|e|f a,b,c,a,e,f
a|b|c|c|d a,b,c,a,c,d
a|a|b|e|f a,a,b,a,e,f
a|a|b|c|d a,a,b,a,c,d
I'm looking at the http://hbase.apache.org/docs/current/api/
Didn't see a multi-get... not sure it would buy you much. My suggestion takes advantage of the fact that HBase puts things in sort order for you.

I think the above solution will work in the general sense, however what's the max width of the row key?
-Mike

> From: [EMAIL PROTECTED]
> To: [EMAIL PROTECTED]
> Date: Tue, 31 May 2011 11:42:27 -0400
> Subject: RE: How to efficiently join HBase tables?
>
> Eran's observation was that a join is solvable in a Mapper via lookups on a 2nd HBase table, but it might not be that efficient if the lookups are 1 by 1.  I agree with that.
>
> My suggestion was to use multi-Get for the lookups instead.  So you'd hold onto a batch of records in the Mapper and then the batch size is filled, then you do the lookups (and then any required emitting, etc.).  
>
>
>
> -----Original Message-----
> From: Michael Segel [mailto:[EMAIL PROTECTED]]
> Sent: Tuesday, May 31, 2011 10:56 AM
> To: [EMAIL PROTECTED]
> Subject: RE: How to efficiently join HBase tables?
>
>
> Maybe I'm missing something... but this isn't a hard problem to solve.
>
> Eran wants to join two tables.
> If we look at an SQL Statement...
>
> SELECT A.*, B.*
> FROM A, B
> WHERE A.1 = B.1
> AND  A.2 = B.2
> AND  A.3 = xxx
> AND A.4 = yyy
> AND B.45 = zzz
>
> Or something along those lines.
>
> So what you're essentially doing is saying I want to take a subset of data from table A, and a subset of data from table B and join them on the values in columns 1 and 2.
> Table A's data will be filtered on columns 3 and 4 and B's data will be filtered on column 45. NOTE: since you don't know the relationship of the column names to either table, you're safer in writing tableA|column_name and tableB|column_name to your temp table.
>
> So if you create a temp table FOO where the key is column 1 and column 2 (column1|column2) then when you walk through the subsets adding them to the temp table, you will get the end result automatically.
>
> Then you can output your hbase temp table and then truncate the table.
     

The Cartesian product often makes an honest-to-god join not such a good idea
on large data.  The common alternative is co-group
which is basically like doing the hard work of the join, but involves
stopping just before emitting the cartesian product.  This allows
you to inject whatever cleverness you need at this point.

Common kinds of cleverness include down-sampling of problematically large
sets of candidates.

On Tue, May 31, 2011 at 11:56 AM, Michael Segel
<[EMAIL PROTECTED]>wrote:

> So the underlying problem that the OP was trying to solve was how to join
> two tables from HBase.
> Unfortunately I goofed.
> I gave a quick and dirty solution that is a bit incomplete. They row key in
> the temp table has to be unique and I forgot about the Cartesian
> product. So my solution wouldn't work in the general case.
>

For my need I don't really need the general case, but even if I did I think
it can probably be done simpler.
The main problem is getting the data from both tables into the same MR job,
without resorting to lookups. So without the theoretical
MutliTableInputFormat, I could just copy all the data from both tables into
a temp table, just append the source table name to the row keys to make sure
there are no conflicts. When all the data from both tables is in the same
temp table, run a MR job. For each row the mapper should emit a key which is
composed of all the values of the join fields in that row (the value can be
emitted as is). This will cause all the rows from both tables, with same
join field values to arrive at the reducer together. The reducer could then
iterate over them and produce the Cartesian product as needed.

I still don't like having to copy all the data into a temp table just
because I can't feed two tables into the MR job.

As Jason Rutherglen mentioned above, Hive can do joins. I don't know if it
can do them for HBase and it will not suit my needs, but it would be
interesting to know how is it doing them, if anyone knows.

-eran

On Tue, May 31, 2011 at 22:02, Ted Dunning <[EMAIL PROTECTED]> wrote:

> The Cartesian product often makes an honest-to-god join not such a good
> idea
> on large data.  The common alternative is co-group
> which is basically like doing the hard work of the join, but involves
> stopping just before emitting the cartesian product.  This allows
> you to inject whatever cleverness you need at this point.
>
> Common kinds of cleverness include down-sampling of problematically large
> sets of candidates.
>
> On Tue, May 31, 2011 at 11:56 AM, Michael Segel
> <[EMAIL PROTECTED]>wrote:
>
> > So the underlying problem that the OP was trying to solve was how to join
> > two tables from HBase.
> > Unfortunately I goofed.
> > I gave a quick and dirty solution that is a bit incomplete. They row key
> in
> > the temp table has to be unique and I forgot about the Cartesian
> > product. So my solution wouldn't work in the general case.
> >
>

Re:  " Didn't see a multi-get... "

This is what I'm talking about...  http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/client/HTable.html#get%28java.util.List%29

re:  " not sure it would buy you much."

Let's say you did these in groups of 500.  Although the reads still obviously need to be performed on the RegionServer, it would cut the total RS calls down by a factor of 500, which is not insignificant.

No matter what approach is taken, this won't be cheap.

As Jean Girard (Talladega Nights) would say, "That is a fair compromise, no?"
-----Original Message-----
From: Michael Segel [mailto:[EMAIL PROTECTED]]
Sent: Tuesday, May 31, 2011 2:56 PM
To: [EMAIL PROTECTED]
Subject: RE: How to efficiently join HBase tables?
Doug,

I read the OP's post as the following:
"> Hi,
> I need to join two HBase tables.
The obvious way is to use a M/R job for that.
The problem is that the few references to that question I found recommend pulling one table to the mapper and then do a lookup for the referred row in the second table.
"

So the underlying problem that the OP was trying to solve was how to join two tables from HBase.
Unfortunately I goofed.
I gave a quick and dirty solution that is a bit incomplete. They row key in the temp table has to be unique and I forgot about the Cartesian product. So my solution wouldn't work in the general case.

If you wanted to do this in the general case...

Map job 1 selects data from the first table and then writes it to tempA where the key is the join column(s) in order, followed by all of the rows. (This is a wide key) Also the map job returns a count of all of the rows in tempA

Map job 2 selects data from the second table again writing the entire return set as the key so that each row is also unique.

Now take the smaller of two sets and run a map reduce job.
For each row, you'd want to create a scan of the other table, using a start and stop row based on the value in the joined columns.
Write out that row and the corresponding row from the scan in to your result set... but here I think you just need a unique key if you're just dumping this to file, otherwise you will want to write the entire record as your key and each column as your value.

This will account for the Cartesian product in the join.

So it will look something like this:
Tab A
K, 1,2,3
   a,b,c
   a,a,b
   b,b,a
   b,c,a
   c,b,a

Tab B
K, 1,2,3
   a,e,f
   a,c,d
   b,z,z

SELECT *
FROM A, B
WHERE A.1 = B.1
AND   A.1 = a
Tab A
K 1,2,3
a|b|c a,b,c
a|a|b a,a,b

Tab B
a|e|f a,e,f
a|c|d a,c,d

So output =
K 1,2,3,4,5,6  
a|b|c|e|f a,b,c,a,e,f
a|b|c|c|d a,b,c,a,c,d
a|a|b|e|f a,a,b,a,e,f
a|a|b|c|d a,a,b,a,c,d
I'm looking at the http://hbase.apache.org/docs/current/api/
Didn't see a multi-get... not sure it would buy you much. My suggestion takes advantage of the fact that HBase puts things in sort order for you.

I think the above solution will work in the general sense, however what's the max width of the row key?
-Mike

> From: [EMAIL PROTECTED]
> To: [EMAIL PROTECTED]
> Date: Tue, 31 May 2011 11:42:27 -0400
> Subject: RE: How to efficiently join HBase tables?
>
> Eran's observation was that a join is solvable in a Mapper via lookups on a 2nd HBase table, but it might not be that efficient if the lookups are 1 by 1.  I agree with that.
>
> My suggestion was to use multi-Get for the lookups instead.  So you'd hold onto a batch of records in the Mapper and then the batch size is filled, then you do the lookups (and then any required emitting, etc.).  
>
>
>
> -----Original Message-----
> From: Michael Segel [mailto:[EMAIL PROTECTED]]
> Sent: Tuesday, May 31, 2011 10:56 AM
> To: [EMAIL PROTECTED]
> Subject: RE: How to efficiently join HBase tables?
>
>
> Maybe I'm missing something... but this isn't a hard problem to solve.
>
> Eran wants to join two tables.
> If we look at an SQL Statement...
>
> SELECT A.*, B.*
> FROM A, B
> WHERE A.1 = B.1
> AND  A.2 = B.2
> AND  A.3 = xxx
     

Eran,

As I said... if you want to do relational database work, you should use a relational database.

The big problem with HBase is that outside of the key, you don't have indexes.
You asked a very general question and we have to assume a general case when we are looking at creating a solution.
Since you don't have the ability to index your fields and join data, you have to assume that you are going to be doing a full table scans.
So rather than go row by row and then doing full table scans, for each row found in one table, the idea is to first get your data from each table separately.
I am again assuming that your subset from Table A is going to be a fraction of the data found in table A and that your subset from table B is going to be a fraction of the data you're getting from table B.

So after your initial filtered scan on each table, you could write them out to sequence files, however, they won't be written in sort order without first doing a reduce job.
(Reduce job will take some time.)

Since you already have a database, writing them to a temp table would probably be faster. (I guess you could write join column(s) | uuid to get uniqueness which would be smaller than trying to cat all of the columns and still be written in 'sort order')

You mentioned that you are afraid about the time it takes for table splits. You could pre-split the table to help alleviate this, but I'm presenting a white board solution that you need to flesh out.

So now you have your temp table of millions of rows where the row keys are the join key with a uuid appended to give uniqueness and its in some sort of sort order. (meaning that all of the keys are grouped.

Now in the actual join, you're Iterating on one temp table, doing a limited scan of the second table.

Its pretty generic.

HTH

-Mike

> From: [EMAIL PROTECTED]
> Date: Tue, 31 May 2011 21:42:58 +0300
> Subject: Re: How to efficiently join HBase tables?
> To: [EMAIL PROTECTED]
>
> Thanks everyone for the great feedback. I'll try to address all the
> suggestions.
>
> My data sets go between large and very large. One is in the order of many
> billions of rows, although the input for a typical MR job will be in the
> hundreds of millions, the second table is in the tens of millions. I doubt a
> SQL DB will handle this kind of a join in a reasonable manner.
>
> Doing batched lookup will indeed be more efficient than one by one but it
> will require the mapper to manage a local state between multiple calls,
> which is something I don't really like doing, and worse, it doesn't really
> solve the lookup problem it only moves it one tier lower. Instead of the
> mapper having to do all thouse random lookups, now HBase itself will have to
> do them. Granted it is more efficient than individual lookup API calls but
> it is not nearly as efficient as doing sequential reads.
>
> Finally, the temp table method, that will work but again, I suspect it will
> be a lot less efficient than the sequence files Hadoop would generate. The
> join output is expected to be in the tens of millions of rows, each with
> multiple columns. From some tests I've done, writing this number of rows to
> a clean table starts out very slowly and takes a lot of time to ramp up as
> the regions begin to split and move around the cluster. I should say that
> the output of this join is just the input for another MR job, so it would
> really be just a temp table and not something that would be useful after
> that.
>
> I should also say that I have looked into eliminating the lookup altogether
> by resolving the data from the second table before the rows are inserted to
> the main table, kind of denormalization, but that would introduce an
> unacceptable latency to a very high volume process.
>
> Still looking for other ideas.
>
> -eran
>
>
>
> On Tue, May 31, 2011 at 18:42, Doug Meil <[EMAIL PROTECTED]>wrote:
>
> > Eran's observation was that a join is solvable in a Mapper via lookups on a
> > 2nd HBase table, but it might not be that efficient if the lookups are 1 by
     

Your mapper can tell which file is being read and add source tags to the
data records.

The reducer can do the cartesian product (if you really need that).

On Tue, May 31, 2011 at 12:19 PM, Eran Kutner <[EMAIL PROTECTED]> wrote:

> For my need I don't really need the general case, but even if I did I think
> it can probably be done simpler.
> The main problem is getting the data from both tables into the same MR job,
> without resorting to lookups. So without the theoretical
> MutliTableInputFormat, I could just copy all the data from both tables into
> a temp table, just append the source table name to the row keys to make
> sure
> there are no conflicts. When all the data from both tables is in the same
> temp table, run a MR job. For each row the mapper should emit a key which
> is
> composed of all the values of the join fields in that row (the value can be
> emitted as is). This will cause all the rows from both tables, with same
> join field values to arrive at the reducer together. The reducer could then
> iterate over them and produce the Cartesian product as needed.
>
> I still don't like having to copy all the data into a temp table just
> because I can't feed two tables into the MR job.
>
> As Jason Rutherglen mentioned above, Hive can do joins. I don't know if it
> can do them for HBase and it will not suit my needs, but it would be
> interesting to know how is it doing them, if anyone knows.
>
> -eran
>
>
>
> On Tue, May 31, 2011 at 22:02, Ted Dunning <[EMAIL PROTECTED]> wrote:
>
> > The Cartesian product often makes an honest-to-god join not such a good
> > idea
> > on large data.  The common alternative is co-group
> > which is basically like doing the hard work of the join, but involves
> > stopping just before emitting the cartesian product.  This allows
> > you to inject whatever cleverness you need at this point.
> >
> > Common kinds of cleverness include down-sampling of problematically large
> > sets of candidates.
> >
> > On Tue, May 31, 2011 at 11:56 AM, Michael Segel
> > <[EMAIL PROTECTED]>wrote:
> >
> > > So the underlying problem that the OP was trying to solve was how to
> join
> > > two tables from HBase.
> > > Unfortunately I goofed.
> > > I gave a quick and dirty solution that is a bit incomplete. They row
> key
> > in
> > > the temp table has to be unique and I forgot about the Cartesian
> > > product. So my solution wouldn't work in the general case.
> > >
> >
>

> From: [EMAIL PROTECTED]
> To: [EMAIL PROTECTED]
> Date: Tue, 31 May 2011 15:39:14 -0400
> Subject: RE: How to efficiently join HBase tables?
>
> Re:  " Didn't see a multi-get... "
>
> This is what I'm talking about...  http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/client/HTable.html#get%28java.util.List%29
>
> re:  " not sure it would buy you much."
>
> Let's say you did these in groups of 500.  Although the reads still obviously need to be performed on the RegionServer, it would cut the total RS calls down by a factor of 500, which is not insignificant.
>
> No matter what approach is taken, this won't be cheap.
>
> As Jean Girard (Talladega Nights) would say, "That is a fair compromise, no?"
>
>

AH Ricky Bobby! :-P

Uhm silly question... how do you use the get(List<Get> gets) if you don't know the row key?

:-)

You're assuming that you're fetching against the row key. I'm assuming that your joining on a foreign key (at best) where the row key isn't known, which is why I'm suggesting sequential scans based on the join criteria.

Its a sad day in Ohio when Tressell resigns. :-(

-Mike
     

On Tue, May 31, 2011 at 3:19 PM, Eran Kutner <[EMAIL PROTECTED]> wrote:

> For my need I don't really need the general case, but even if I did I think
> it can probably be done simpler.
> The main problem is getting the data from both tables into the same MR job,
> without resorting to lookups. So without the theoretical
> MutliTableInputFormat, I could just copy all the data from both tables into
> a temp table, just append the source table name to the row keys to make
> sure
> there are no conflicts. When all the data from both tables is in the same
> temp table, run a MR job. For each row the mapper should emit a key which
> is
> composed of all the values of the join fields in that row (the value can be
> emitted as is). This will cause all the rows from both tables, with same
> join field values to arrive at the reducer together. The reducer could then
> iterate over them and produce the Cartesian product as needed.
>
> I still don't like having to copy all the data into a temp table just
> because I can't feed two tables into the MR job.
>

Loading the smaller table in memory is called a map join, versus a
reduce-side join (a.k.a. common join). One reason to prefer a map join is
you avoid the shuffle phase which potentially involves several trips to disk
for the intermediate records due to spills, and also once through the
network to get each intermediate KV pair to the right reducer. With a map
join, everything is local, except for the part where you load the small
table.
>
> As Jason Rutherglen mentioned above, Hive can do joins. I don't know if it
> can do them for HBase and it will not suit my needs, but it would be
> interesting to know how is it doing them, if anyone knows.
>

The Hive-HBase integration allows you to create Hive tables that are backed
by HBase. You can do joins on those tables (and also with standard Hive
tables). It might be worth trying out in your case as it lets you easily see
the load characteristics and the job runtime without much coding investment.

There are probably some specific optimizations that can be applied to your
situation, but it's hard to say without knowing your use-case.

Regards,

- Patrick
> -eran
>
>
>
> On Tue, May 31, 2011 at 22:02, Ted Dunning <[EMAIL PROTECTED]> wrote:
>
> > The Cartesian product often makes an honest-to-god join not such a good
> > idea
> > on large data.  The common alternative is co-group
> > which is basically like doing the hard work of the join, but involves
> > stopping just before emitting the cartesian product.  This allows
> > you to inject whatever cleverness you need at this point.
> >
> > Common kinds of cleverness include down-sampling of problematically large
> > sets of candidates.
> >
> > On Tue, May 31, 2011 at 11:56 AM, Michael Segel
> > <[EMAIL PROTECTED]>wrote:
> >
> > > So the underlying problem that the OP was trying to solve was how to
> join
> > > two tables from HBase.
> > > Unfortunately I goofed.
> > > I gave a quick and dirty solution that is a bit incomplete. They row
> key
> > in
> > > the temp table has to be unique and I forgot about the Cartesian
> > > product. So my solution wouldn't work in the general case.
> > >
> >
>

> The Hive-HBase integration allows you to create Hive tables that are backed
> by HBase

In addition, HBase can be made to go faster for MapReduce jobs, if the
HFile's could be used directly in HDFS, rather than proxying through
the RegionServer.

I'd imagine that join operations do not require realtime-ness, and so
faster batch jobs using Hive -> frozen HBase files in HDFS could be
the optimal way to go?

On Tue, May 31, 2011 at 1:41 PM, Patrick Angeles <[EMAIL PROTECTED]> wrote:
> On Tue, May 31, 2011 at 3:19 PM, Eran Kutner <[EMAIL PROTECTED]> wrote:
>
>> For my need I don't really need the general case, but even if I did I think
>> it can probably be done simpler.
>> The main problem is getting the data from both tables into the same MR job,
>> without resorting to lookups. So without the theoretical
>> MutliTableInputFormat, I could just copy all the data from both tables into
>> a temp table, just append the source table name to the row keys to make
>> sure
>> there are no conflicts. When all the data from both tables is in the same
>> temp table, run a MR job. For each row the mapper should emit a key which
>> is
>> composed of all the values of the join fields in that row (the value can be
>> emitted as is). This will cause all the rows from both tables, with same
>> join field values to arrive at the reducer together. The reducer could then
>> iterate over them and produce the Cartesian product as needed.
>>
>> I still don't like having to copy all the data into a temp table just
>> because I can't feed two tables into the MR job.
>>
>
> Loading the smaller table in memory is called a map join, versus a
> reduce-side join (a.k.a. common join). One reason to prefer a map join is
> you avoid the shuffle phase which potentially involves several trips to disk
> for the intermediate records due to spills, and also once through the
> network to get each intermediate KV pair to the right reducer. With a map
> join, everything is local, except for the part where you load the small
> table.
>
>
>>
>> As Jason Rutherglen mentioned above, Hive can do joins. I don't know if it
>> can do them for HBase and it will not suit my needs, but it would be
>> interesting to know how is it doing them, if anyone knows.
>>
>
> The Hive-HBase integration allows you to create Hive tables that are backed
> by HBase. You can do joins on those tables (and also with standard Hive
> tables). It might be worth trying out in your case as it lets you easily see
> the load characteristics and the job runtime without much coding investment.
>
> There are probably some specific optimizations that can be applied to your
> situation, but it's hard to say without knowing your use-case.
>
> Regards,
>
> - Patrick
>
>
>> -eran
>>
>>
>>
>> On Tue, May 31, 2011 at 22:02, Ted Dunning <[EMAIL PROTECTED]> wrote:
>>
>> > The Cartesian product often makes an honest-to-god join not such a good
>> > idea
>> > on large data.  The common alternative is co-group
>> > which is basically like doing the hard work of the join, but involves
>> > stopping just before emitting the cartesian product.  This allows
>> > you to inject whatever cleverness you need at this point.
>> >
>> > Common kinds of cleverness include down-sampling of problematically large
>> > sets of candidates.
>> >
>> > On Tue, May 31, 2011 at 11:56 AM, Michael Segel
>> > <[EMAIL PROTECTED]>wrote:
>> >
>> > > So the underlying problem that the OP was trying to solve was how to
>> join
>> > > two tables from HBase.
>> > > Unfortunately I goofed.
>> > > I gave a quick and dirty solution that is a bit incomplete. They row
>> key
>> > in
>> > > the temp table has to be unique and I forgot about the Cartesian
>> > > product. So my solution wouldn't work in the general case.
>> > >
>> >
>>
>

We use Pig to join HBase tables using HBaseStorage which has worked well. If
you're using HBase >= 0.89 you'll need to build from the trunk or the Pig
0.8 branch.
On Tue, May 31, 2011 at 5:18 PM, Jason Rutherglen <
[EMAIL PROTECTED]> wrote:

> > The Hive-HBase integration allows you to create Hive tables that are
> backed
> > by HBase
>
> In addition, HBase can be made to go faster for MapReduce jobs, if the
> HFile's could be used directly in HDFS, rather than proxying through
> the RegionServer.
>
> I'd imagine that join operations do not require realtime-ness, and so
> faster batch jobs using Hive -> frozen HBase files in HDFS could be
> the optimal way to go?
>
> On Tue, May 31, 2011 at 1:41 PM, Patrick Angeles <[EMAIL PROTECTED]>
> wrote:
> > On Tue, May 31, 2011 at 3:19 PM, Eran Kutner <[EMAIL PROTECTED]> wrote:
> >
> >> For my need I don't really need the general case, but even if I did I
> think
> >> it can probably be done simpler.
> >> The main problem is getting the data from both tables into the same MR
> job,
> >> without resorting to lookups. So without the theoretical
> >> MutliTableInputFormat, I could just copy all the data from both tables
> into
> >> a temp table, just append the source table name to the row keys to make
> >> sure
> >> there are no conflicts. When all the data from both tables is in the
> same
> >> temp table, run a MR job. For each row the mapper should emit a key
> which
> >> is
> >> composed of all the values of the join fields in that row (the value can
> be
> >> emitted as is). This will cause all the rows from both tables, with same
> >> join field values to arrive at the reducer together. The reducer could
> then
> >> iterate over them and produce the Cartesian product as needed.
> >>
> >> I still don't like having to copy all the data into a temp table just
> >> because I can't feed two tables into the MR job.
> >>
> >
> > Loading the smaller table in memory is called a map join, versus a
> > reduce-side join (a.k.a. common join). One reason to prefer a map join is
> > you avoid the shuffle phase which potentially involves several trips to
> disk
> > for the intermediate records due to spills, and also once through the
> > network to get each intermediate KV pair to the right reducer. With a map
> > join, everything is local, except for the part where you load the small
> > table.
> >
> >
> >>
> >> As Jason Rutherglen mentioned above, Hive can do joins. I don't know if
> it
> >> can do them for HBase and it will not suit my needs, but it would be
> >> interesting to know how is it doing them, if anyone knows.
> >>
> >
> > The Hive-HBase integration allows you to create Hive tables that are
> backed
> > by HBase. You can do joins on those tables (and also with standard Hive
> > tables). It might be worth trying out in your case as it lets you easily
> see
> > the load characteristics and the job runtime without much coding
> investment.
> >
> > There are probably some specific optimizations that can be applied to
> your
> > situation, but it's hard to say without knowing your use-case.
> >
> > Regards,
> >
> > - Patrick
> >
> >
> >> -eran
> >>
> >>
> >>
> >> On Tue, May 31, 2011 at 22:02, Ted Dunning <[EMAIL PROTECTED]>
> wrote:
> >>
> >> > The Cartesian product often makes an honest-to-god join not such a
> good
> >> > idea
> >> > on large data.  The common alternative is co-group
> >> > which is basically like doing the hard work of the join, but involves
> >> > stopping just before emitting the cartesian product.  This allows
> >> > you to inject whatever cleverness you need at this point.
> >> >
> >> > Common kinds of cleverness include down-sampling of problematically
> large
> >> > sets of candidates.
> >> >
> >> > On Tue, May 31, 2011 at 11:56 AM, Michael Segel
> >> > <[EMAIL PROTECTED]>wrote:
> >> >
> >> > > So the underlying problem that the OP was trying to solve was how to
> >> join
> >> > > two tables from HBase.
> >> > > Unfortunately I goofed.
> >> > > I gave a quick and dirty solution that is a bit incomplete. They row

> I'd imagine that join operations do not require realtime-ness, and so
> faster batch jobs using Hive -> frozen HBase files in HDFS could be
> the optimal way to go?

In addition to lessening the load on the perhaps live RegionServer.
There's no Jira for this, I'm tempted to open one.

On Tue, May 31, 2011 at 5:18 PM, Jason Rutherglen
<[EMAIL PROTECTED]> wrote:
>> The Hive-HBase integration allows you to create Hive tables that are backed
>> by HBase
>
> In addition, HBase can be made to go faster for MapReduce jobs, if the
> HFile's could be used directly in HDFS, rather than proxying through
> the RegionServer.
>
> I'd imagine that join operations do not require realtime-ness, and so
> faster batch jobs using Hive -> frozen HBase files in HDFS could be
> the optimal way to go?
>
> On Tue, May 31, 2011 at 1:41 PM, Patrick Angeles <[EMAIL PROTECTED]> wrote:
>> On Tue, May 31, 2011 at 3:19 PM, Eran Kutner <[EMAIL PROTECTED]> wrote:
>>
>>> For my need I don't really need the general case, but even if I did I think
>>> it can probably be done simpler.
>>> The main problem is getting the data from both tables into the same MR job,
>>> without resorting to lookups. So without the theoretical
>>> MutliTableInputFormat, I could just copy all the data from both tables into
>>> a temp table, just append the source table name to the row keys to make
>>> sure
>>> there are no conflicts. When all the data from both tables is in the same
>>> temp table, run a MR job. For each row the mapper should emit a key which
>>> is
>>> composed of all the values of the join fields in that row (the value can be
>>> emitted as is). This will cause all the rows from both tables, with same
>>> join field values to arrive at the reducer together. The reducer could then
>>> iterate over them and produce the Cartesian product as needed.
>>>
>>> I still don't like having to copy all the data into a temp table just
>>> because I can't feed two tables into the MR job.
>>>
>>
>> Loading the smaller table in memory is called a map join, versus a
>> reduce-side join (a.k.a. common join). One reason to prefer a map join is
>> you avoid the shuffle phase which potentially involves several trips to disk
>> for the intermediate records due to spills, and also once through the
>> network to get each intermediate KV pair to the right reducer. With a map
>> join, everything is local, except for the part where you load the small
>> table.
>>
>>
>>>
>>> As Jason Rutherglen mentioned above, Hive can do joins. I don't know if it
>>> can do them for HBase and it will not suit my needs, but it would be
>>> interesting to know how is it doing them, if anyone knows.
>>>
>>
>> The Hive-HBase integration allows you to create Hive tables that are backed
>> by HBase. You can do joins on those tables (and also with standard Hive
>> tables). It might be worth trying out in your case as it lets you easily see
>> the load characteristics and the job runtime without much coding investment.
>>
>> There are probably some specific optimizations that can be applied to your
>> situation, but it's hard to say without knowing your use-case.
>>
>> Regards,
>>
>> - Patrick
>>
>>
>>> -eran
>>>
>>>
>>>
>>> On Tue, May 31, 2011 at 22:02, Ted Dunning <[EMAIL PROTECTED]> wrote:
>>>
>>> > The Cartesian product often makes an honest-to-god join not such a good
>>> > idea
>>> > on large data.  The common alternative is co-group
>>> > which is basically like doing the hard work of the join, but involves
>>> > stopping just before emitting the cartesian product.  This allows
>>> > you to inject whatever cleverness you need at this point.
>>> >
>>> > Common kinds of cleverness include down-sampling of problematically large
>>> > sets of candidates.
>>> >
>>> > On Tue, May 31, 2011 at 11:56 AM, Michael Segel
>>> > <[EMAIL PROTECTED]>wrote:
>>> >
>>> > > So the underlying problem that the OP was trying to solve was how to
>>> join
>>> > > two tables from HBase.
>>> > > Unfortunately I goofed.

Thanks everyone for all the helpful insights!

-eran

On Wed, Jun 1, 2011 at 03:41, Jason Rutherglen
<[EMAIL PROTECTED]>wrote:

> > I'd imagine that join operations do not require realtime-ness, and so
> > faster batch jobs using Hive -> frozen HBase files in HDFS could be
> > the optimal way to go?
>
> In addition to lessening the load on the perhaps live RegionServer.
> There's no Jira for this, I'm tempted to open one.
>
> On Tue, May 31, 2011 at 5:18 PM, Jason Rutherglen
> <[EMAIL PROTECTED]> wrote:
> >> The Hive-HBase integration allows you to create Hive tables that are
> backed
> >> by HBase
> >
> > In addition, HBase can be made to go faster for MapReduce jobs, if the
> > HFile's could be used directly in HDFS, rather than proxying through
> > the RegionServer.
> >
> > I'd imagine that join operations do not require realtime-ness, and so
> > faster batch jobs using Hive -> frozen HBase files in HDFS could be
> > the optimal way to go?
> >
> > On Tue, May 31, 2011 at 1:41 PM, Patrick Angeles <[EMAIL PROTECTED]>
> wrote:
> >> On Tue, May 31, 2011 at 3:19 PM, Eran Kutner <[EMAIL PROTECTED]> wrote:
> >>
> >>> For my need I don't really need the general case, but even if I did I
> think
> >>> it can probably be done simpler.
> >>> The main problem is getting the data from both tables into the same MR
> job,
> >>> without resorting to lookups. So without the theoretical
> >>> MutliTableInputFormat, I could just copy all the data from both tables
> into
> >>> a temp table, just append the source table name to the row keys to make
> >>> sure
> >>> there are no conflicts. When all the data from both tables is in the
> same
> >>> temp table, run a MR job. For each row the mapper should emit a key
> which
> >>> is
> >>> composed of all the values of the join fields in that row (the value
> can be
> >>> emitted as is). This will cause all the rows from both tables, with
> same
> >>> join field values to arrive at the reducer together. The reducer could
> then
> >>> iterate over them and produce the Cartesian product as needed.
> >>>
> >>> I still don't like having to copy all the data into a temp table just
> >>> because I can't feed two tables into the MR job.
> >>>
> >>
> >> Loading the smaller table in memory is called a map join, versus a
> >> reduce-side join (a.k.a. common join). One reason to prefer a map join
> is
> >> you avoid the shuffle phase which potentially involves several trips to
> disk
> >> for the intermediate records due to spills, and also once through the
> >> network to get each intermediate KV pair to the right reducer. With a
> map
> >> join, everything is local, except for the part where you load the small
> >> table.
> >>
> >>
> >>>
> >>> As Jason Rutherglen mentioned above, Hive can do joins. I don't know if
> it
> >>> can do them for HBase and it will not suit my needs, but it would be
> >>> interesting to know how is it doing them, if anyone knows.
> >>>
> >>
> >> The Hive-HBase integration allows you to create Hive tables that are
> backed
> >> by HBase. You can do joins on those tables (and also with standard Hive
> >> tables). It might be worth trying out in your case as it lets you easily
> see
> >> the load characteristics and the job runtime without much coding
> investment.
> >>
> >> There are probably some specific optimizations that can be applied to
> your
> >> situation, but it's hard to say without knowing your use-case.
> >>
> >> Regards,
> >>
> >> - Patrick
> >>
> >>
> >>> -eran
> >>>
> >>>
> >>>
> >>> On Tue, May 31, 2011 at 22:02, Ted Dunning <[EMAIL PROTECTED]>
> wrote:
> >>>
> >>> > The Cartesian product often makes an honest-to-god join not such a
> good
> >>> > idea
> >>> > on large data.  The common alternative is co-group
> >>> > which is basically like doing the hard work of the join, but involves
> >>> > stopping just before emitting the cartesian product.  This allows
> >>> > you to inject whatever cleverness you need at this point.
> >>> >
> >>> > Common kinds of cleverness include down-sampling of problematically

Hi Jason,

This was discussed in the past, using the HFileInputFormat. The issue
is that you somehow need to flush all in-memory data *and* perform a
major compaction - or else you would need all the logic of the
ColumnTracker in the HFIF. Since that needs to scan all storage files
in parallel to achieve its job, the MR task would not really be able
to use the same approach.

Running a major compaction creates a lot of churn, so it is
questionable what the outcome is. The files do have a flag if they
were made by a major compaction, so you scan only those and ignore the
newer ones - but then you are trailing, and you still do not handle
delete markers/updates in newer files. No easy feat.

Lars

On Wed, Jun 1, 2011 at 2:41 AM, Jason Rutherglen
<[EMAIL PROTECTED]> wrote:
>> I'd imagine that join operations do not require realtime-ness, and so
>> faster batch jobs using Hive -> frozen HBase files in HDFS could be
>> the optimal way to go?
>
> In addition to lessening the load on the perhaps live RegionServer.
> There's no Jira for this, I'm tempted to open one.
>
> On Tue, May 31, 2011 at 5:18 PM, Jason Rutherglen
> <[EMAIL PROTECTED]> wrote:
>>> The Hive-HBase integration allows you to create Hive tables that are backed
>>> by HBase
>>
>> In addition, HBase can be made to go faster for MapReduce jobs, if the
>> HFile's could be used directly in HDFS, rather than proxying through
>> the RegionServer.
>>
>> I'd imagine that join operations do not require realtime-ness, and so
>> faster batch jobs using Hive -> frozen HBase files in HDFS could be
>> the optimal way to go?
>>
>> On Tue, May 31, 2011 at 1:41 PM, Patrick Angeles <[EMAIL PROTECTED]> wrote:
>>> On Tue, May 31, 2011 at 3:19 PM, Eran Kutner <[EMAIL PROTECTED]> wrote:
>>>
>>>> For my need I don't really need the general case, but even if I did I think
>>>> it can probably be done simpler.
>>>> The main problem is getting the data from both tables into the same MR job,
>>>> without resorting to lookups. So without the theoretical
>>>> MutliTableInputFormat, I could just copy all the data from both tables into
>>>> a temp table, just append the source table name to the row keys to make
>>>> sure
>>>> there are no conflicts. When all the data from both tables is in the same
>>>> temp table, run a MR job. For each row the mapper should emit a key which
>>>> is
>>>> composed of all the values of the join fields in that row (the value can be
>>>> emitted as is). This will cause all the rows from both tables, with same
>>>> join field values to arrive at the reducer together. The reducer could then
>>>> iterate over them and produce the Cartesian product as needed.
>>>>
>>>> I still don't like having to copy all the data into a temp table just
>>>> because I can't feed two tables into the MR job.
>>>>
>>>
>>> Loading the smaller table in memory is called a map join, versus a
>>> reduce-side join (a.k.a. common join). One reason to prefer a map join is
>>> you avoid the shuffle phase which potentially involves several trips to disk
>>> for the intermediate records due to spills, and also once through the
>>> network to get each intermediate KV pair to the right reducer. With a map
>>> join, everything is local, except for the part where you load the small
>>> table.
>>>
>>>
>>>>
>>>> As Jason Rutherglen mentioned above, Hive can do joins. I don't know if it
>>>> can do them for HBase and it will not suit my needs, but it would be
>>>> interesting to know how is it doing them, if anyone knows.
>>>>
>>>
>>> The Hive-HBase integration allows you to create Hive tables that are backed
>>> by HBase. You can do joins on those tables (and also with standard Hive
>>> tables). It might be worth trying out in your case as it lets you easily see
>>> the load characteristics and the job runtime without much coding investment.
>>>
>>> There are probably some specific optimizations that can be applied to your
>>> situation, but it's hard to say without knowing your use-case.
>

> you somehow need to flush all in-memory data *and* perform a
> major compaction

This makes sense.  Without compaction the linear HDFS scan isn't
possible.  I suppose one could compact 'offline' in a different Map
Reduce job.  However that would have it's own issues.

> The files do have a flag if they were made by a major compaction,
> so you scan only those and ignore the newer ones - but then you are trailing

This could be ok in many cases.  The key would be to create a sync'd
cut off point enabling a frozen point-in-time 'view' of the data.  I'm
not sure how that would be implemented.

On Wed, Jun 1, 2011 at 6:54 AM, Lars George <[EMAIL PROTECTED]> wrote:
> Hi Jason,
>
> This was discussed in the past, using the HFileInputFormat. The issue
> is that you somehow need to flush all in-memory data *and* perform a
> major compaction - or else you would need all the logic of the
> ColumnTracker in the HFIF. Since that needs to scan all storage files
> in parallel to achieve its job, the MR task would not really be able
> to use the same approach.
>
> Running a major compaction creates a lot of churn, so it is
> questionable what the outcome is. The files do have a flag if they
> were made by a major compaction, so you scan only those and ignore the
> newer ones - but then you are trailing, and you still do not handle
> delete markers/updates in newer files. No easy feat.
>
> Lars
>
> On Wed, Jun 1, 2011 at 2:41 AM, Jason Rutherglen
> <[EMAIL PROTECTED]> wrote:
>>> I'd imagine that join operations do not require realtime-ness, and so
>>> faster batch jobs using Hive -> frozen HBase files in HDFS could be
>>> the optimal way to go?
>>
>> In addition to lessening the load on the perhaps live RegionServer.
>> There's no Jira for this, I'm tempted to open one.
>>
>> On Tue, May 31, 2011 at 5:18 PM, Jason Rutherglen
>> <[EMAIL PROTECTED]> wrote:
>>>> The Hive-HBase integration allows you to create Hive tables that are backed
>>>> by HBase
>>>
>>> In addition, HBase can be made to go faster for MapReduce jobs, if the
>>> HFile's could be used directly in HDFS, rather than proxying through
>>> the RegionServer.
>>>
>>> I'd imagine that join operations do not require realtime-ness, and so
>>> faster batch jobs using Hive -> frozen HBase files in HDFS could be
>>> the optimal way to go?
>>>
>>> On Tue, May 31, 2011 at 1:41 PM, Patrick Angeles <[EMAIL PROTECTED]> wrote:
>>>> On Tue, May 31, 2011 at 3:19 PM, Eran Kutner <[EMAIL PROTECTED]> wrote:
>>>>
>>>>> For my need I don't really need the general case, but even if I did I think
>>>>> it can probably be done simpler.
>>>>> The main problem is getting the data from both tables into the same MR job,
>>>>> without resorting to lookups. So without the theoretical
>>>>> MutliTableInputFormat, I could just copy all the data from both tables into
>>>>> a temp table, just append the source table name to the row keys to make
>>>>> sure
>>>>> there are no conflicts. When all the data from both tables is in the same
>>>>> temp table, run a MR job. For each row the mapper should emit a key which
>>>>> is
>>>>> composed of all the values of the join fields in that row (the value can be
>>>>> emitted as is). This will cause all the rows from both tables, with same
>>>>> join field values to arrive at the reducer together. The reducer could then
>>>>> iterate over them and produce the Cartesian product as needed.
>>>>>
>>>>> I still don't like having to copy all the data into a temp table just
>>>>> because I can't feed two tables into the MR job.
>>>>>
>>>>
>>>> Loading the smaller table in memory is called a map join, versus a
>>>> reduce-side join (a.k.a. common join). One reason to prefer a map join is
>>>> you avoid the shuffle phase which potentially involves several trips to disk
>>>> for the intermediate records due to spills, and also once through the
>>>> network to get each intermediate KV pair to the right reducer. With a map
>>>> join, everything is local, except for the part where you load the small

Not to beat a dead horse, but I thought a bit more about the problem.
If you want to do this all in HBase using a M/R job...

Lets define the following:
SELECT *
FROM A, B
WHERE A.a = B.a
AND     A.b = B.b
AND     A.c = xxx
AND     A.d = yyy
AND     B.e = zzz

Is the sample query.

So our join key is "a|b" because we're matching on columns a and b. (The pipe is to delimit the columns, assuming the columns themselves don't contain pipes...)

Our filters on A are c and d while e is the filter on B.

So we want to do the following:

M/R Map job 1 gets the subset from table A along with a set of unique keys.
M/R Map job 2 gets the subset from table B along with a set of unique keys.
M/R Map job 3 takes either set of unique keys as the input list and you split it based on the number of parallel mappers you want to use.

You have a couple of options on how you want to proceed.
In each Mapper.map() your input is a unique key.
I guess you could create two scanners, one for tempTableA, and one for tempTableB.
It looks like you can get the iterator for each result set, and then for each row in temp table A, you iterate through the result set from temp table B, writing out the joined set.

The only problem is that your result set file isn't in sort order. So I guess you could take the output from this job and reduce it to get it in to sort order.

Option B. Using HDFS files for temp 'tables'.
You can do this... but you would still have to track the unique keys and also sort both the keys and the files which will require a reduce job.
Now this is just my opinion, but if I use HBase, I don't have to worry about using a reducer except to order the final output set.
So I can save the time it takes to do the reduce step. So I have to ask... how much time is spent by HBase in splitting and compacting the temp tables?
Also can't you pre-split the temp table before you use them?

Or am I still missing something?

Note: In this example, you'd have to write an input format that takes a java list object (or something similar) as your input and then you can split it to get it to run in parallel.
Or you could just write this on the client and split the list up and run the join in parallel threads on the client node. Or a single thread which would mean that it would run and output in sort order.

HTH

-Mike

> Date: Wed, 1 Jun 2011 07:47:30 -0700
> Subject: Re: How to efficiently join HBase tables?
> From: [EMAIL PROTECTED]
> To: [EMAIL PROTECTED]
>
> > you somehow need to flush all in-memory data *and* perform a
> > major compaction
>
> This makes sense.  Without compaction the linear HDFS scan isn't
> possible.  I suppose one could compact 'offline' in a different Map
> Reduce job.  However that would have it's own issues.
>
> > The files do have a flag if they were made by a major compaction,
> > so you scan only those and ignore the newer ones - but then you are trailing
>
> This could be ok in many cases.  The key would be to create a sync'd
> cut off point enabling a frozen point-in-time 'view' of the data.  I'm
> not sure how that would be implemented.
>
> On Wed, Jun 1, 2011 at 6:54 AM, Lars George <[EMAIL PROTECTED]> wrote:
> > Hi Jason,
> >
> > This was discussed in the past, using the HFileInputFormat. The issue
> > is that you somehow need to flush all in-memory data *and* perform a
> > major compaction - or else you would need all the logic of the
> > ColumnTracker in the HFIF. Since that needs to scan all storage files
> > in parallel to achieve its job, the MR task would not really be able
> > to use the same approach.
> >
> > Running a major compaction creates a lot of churn, so it is
> > questionable what the outcome is. The files do have a flag if they
> > were made by a major compaction, so you scan only those and ignore the
> > newer ones - but then you are trailing, and you still do not handle
> > delete markers/updates in newer files. No easy feat.
> >
> > Lars
> >
> > On Wed, Jun 1, 2011 at 2:41 AM, Jason Rutherglen
     

Mike, this more or less what I tried to  describe in my initial post, only
you explained it much better.
The problem is that I want to do all of this in one M/R run, not 3 and
without explicit temp tables. If there was only a way to feed both table A
and table B into the M/R job then it could be done.

Let's take your query and assumptions, for example.
So we configure scanner A to return rows where c=xxx and d=yyy
We then configure scanner B to return rows where e=zzz
Now we feed all those rows to the mapper.
For each row the mapper gets it outputs a new key which is "a|b" and the
same value it received, if either one doesn't exist in the row the mapper
doesn't output anything for that row.
The is an implicit "temp table" created at this stage by hadoop.
Now the reducer is run, for every key "a|b" generated by the mapper it would
get one or more value sets, each one representing a row from the original
two tables. For simplicity lets assume we got two rows, one from table A the
other from table B. Now the reducer can combine the two rows and output the
combined row. This will work just the same if there were multiple rows from
each table with the same "a|b" key, in that case the reducer would have to
generate the Cartesian product of all the rows. Outer joins can also be done
this way, in an outer join you only get one row in the reducer for a given
"a|b" key but still generate an output.

-eran

On Fri, Jun 3, 2011 at 00:05, Michael Segel <[EMAIL PROTECTED]>wrote:

>
> Not to beat a dead horse, but I thought a bit more about the problem.
> If you want to do this all in HBase using a M/R job...
>
> Lets define the following:
> SELECT *
> FROM A, B
> WHERE A.a = B.a
> AND     A.b = B.b
> AND     A.c = xxx
> AND     A.d = yyy
> AND     B.e = zzz
>
> Is the sample query.
>
> So our join key is "a|b" because we're matching on columns a and b. (The
> pipe is to delimit the columns, assuming the columns themselves don't
> contain pipes...)
>
> Our filters on A are c and d while e is the filter on B.
>
> So we want to do the following:
>
> M/R Map job 1 gets the subset from table A along with a set of unique keys.
> M/R Map job 2 gets the subset from table B along with a set of unique keys.
> M/R Map job 3 takes either set of unique keys as the input list and you
> split it based on the number of parallel mappers you want to use.
>
> You have a couple of options on how you want to proceed.
> In each Mapper.map() your input is a unique key.
> I guess you could create two scanners, one for tempTableA, and one for
> tempTableB.
> It looks like you can get the iterator for each result set, and then for
> each row in temp table A, you iterate through the result set from temp table
> B, writing out the joined set.
>
> The only problem is that your result set file isn't in sort order. So I
> guess you could take the output from this job and reduce it to get it in to
> sort order.
>
> Option B. Using HDFS files for temp 'tables'.
> You can do this... but you would still have to track the unique keys and
> also sort both the keys and the files which will require a reduce job.
>
>
> Now this is just my opinion, but if I use HBase, I don't have to worry
> about using a reducer except to order the final output set.
> So I can save the time it takes to do the reduce step. So I have to ask...
> how much time is spent by HBase in splitting and compacting the temp tables?
> Also can't you pre-split the temp table before you use them?
>
> Or am I still missing something?
>
> Note: In this example, you'd have to write an input format that takes a
> java list object (or something similar) as your input and then you can split
> it to get it to run in parallel.
> Or you could just write this on the client and split the list up and run
> the join in parallel threads on the client node. Or a single thread which
> would mean that it would run and output in sort order.
>
> HTH
>
> -Mike
>
> > Date: Wed, 1 Jun 2011 07:47:30 -0700
> > Subject: Re: How to efficiently join HBase tables?

So, you all realize the joins have been talked about in the database community for 40 years?  There are two main types of joins:
Nested loops
Hash table

Mike, in his various emails seems to be trying to re-imagine how to implement both types of joins in HBase (which seems like a reasonable goal). I am not exactly sure what Eran is going for here, but it seems like Eran is glossing over a piece.  If you have two scanners for table A and B, then table B needs to be rescanned for every unique part of the join condition in table A.  There are certain ways of improving the efficiency of that: the two most obvious are pushing the selection criteria down to the scans, and scanning all of the same join values from table B at the same time (which requires that Table B's key is the join, or a secondary structure that stores the join values as the primary order).

Dave

-----Original Message-----
From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran Kutner
Sent: Friday, June 03, 2011 12:24 AM
To: [EMAIL PROTECTED]
Subject: Re: How to efficiently join HBase tables?

Mike, this more or less what I tried to  describe in my initial post, only
you explained it much better.
The problem is that I want to do all of this in one M/R run, not 3 and
without explicit temp tables. If there was only a way to feed both table A
and table B into the M/R job then it could be done.

Let's take your query and assumptions, for example.
So we configure scanner A to return rows where c=xxx and d=yyy
We then configure scanner B to return rows where e=zzz
Now we feed all those rows to the mapper.
For each row the mapper gets it outputs a new key which is "a|b" and the
same value it received, if either one doesn't exist in the row the mapper
doesn't output anything for that row.
The is an implicit "temp table" created at this stage by hadoop.
Now the reducer is run, for every key "a|b" generated by the mapper it would
get one or more value sets, each one representing a row from the original
two tables. For simplicity lets assume we got two rows, one from table A the
other from table B. Now the reducer can combine the two rows and output the
combined row. This will work just the same if there were multiple rows from
each table with the same "a|b" key, in that case the reducer would have to
generate the Cartesian product of all the rows. Outer joins can also be done
this way, in an outer join you only get one row in the reducer for a given
"a|b" key but still generate an output.

-eran

On Fri, Jun 3, 2011 at 00:05, Michael Segel <[EMAIL PROTECTED]>wrote:

>
> Not to beat a dead horse, but I thought a bit more about the problem.
> If you want to do this all in HBase using a M/R job...
>
> Lets define the following:
> SELECT *
> FROM A, B
> WHERE A.a = B.a
> AND     A.b = B.b
> AND     A.c = xxx
> AND     A.d = yyy
> AND     B.e = zzz
>
> Is the sample query.
>
> So our join key is "a|b" because we're matching on columns a and b. (The
> pipe is to delimit the columns, assuming the columns themselves don't
> contain pipes...)
>
> Our filters on A are c and d while e is the filter on B.
>
> So we want to do the following:
>
> M/R Map job 1 gets the subset from table A along with a set of unique keys.
> M/R Map job 2 gets the subset from table B along with a set of unique keys.
> M/R Map job 3 takes either set of unique keys as the input list and you
> split it based on the number of parallel mappers you want to use.
>
> You have a couple of options on how you want to proceed.
> In each Mapper.map() your input is a unique key.
> I guess you could create two scanners, one for tempTableA, and one for
> tempTableB.
> It looks like you can get the iterator for each result set, and then for
> each row in temp table A, you iterate through the result set from temp table
> B, writing out the joined set.
>
> The only problem is that your result set file isn't in sort order. So I
> guess you could take the output from this job and reduce it to get it in to

Re:  " So, you all realize the joins have been talked about in the database community for 40 years?"

Great point.  What's old is new!    :-)

My suggested from earlier in the thread was a variant of nested loops by using multi-get in HTable, which would reduce the number of RPC calls.  So it's a "bulk-select nested loops" of sorts (i.e., as opposed to the 1-by-1 lookup of regular nested loops).
-----Original Message-----
From: Buttler, David [mailto:[EMAIL PROTECTED]]
Sent: Monday, June 06, 2011 4:30 PM
To: [EMAIL PROTECTED]
Subject: RE: How to efficiently join HBase tables?

So, you all realize the joins have been talked about in the database community for 40 years?  There are two main types of joins:
Nested loops
Hash table

Mike, in his various emails seems to be trying to re-imagine how to implement both types of joins in HBase (which seems like a reasonable goal). I am not exactly sure what Eran is going for here, but it seems like Eran is glossing over a piece.  If you have two scanners for table A and B, then table B needs to be rescanned for every unique part of the join condition in table A.  There are certain ways of improving the efficiency of that: the two most obvious are pushing the selection criteria down to the scans, and scanning all of the same join values from table B at the same time (which requires that Table B's key is the join, or a secondary structure that stores the join values as the primary order).

Dave

-----Original Message-----
From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran Kutner
Sent: Friday, June 03, 2011 12:24 AM
To: [EMAIL PROTECTED]
Subject: Re: How to efficiently join HBase tables?

Mike, this more or less what I tried to  describe in my initial post, only you explained it much better.
The problem is that I want to do all of this in one M/R run, not 3 and without explicit temp tables. If there was only a way to feed both table A and table B into the M/R job then it could be done.

Let's take your query and assumptions, for example.
So we configure scanner A to return rows where c=xxx and d=yyy We then configure scanner B to return rows where e=zzz Now we feed all those rows to the mapper.
For each row the mapper gets it outputs a new key which is "a|b" and the same value it received, if either one doesn't exist in the row the mapper doesn't output anything for that row.
The is an implicit "temp table" created at this stage by hadoop.
Now the reducer is run, for every key "a|b" generated by the mapper it would get one or more value sets, each one representing a row from the original two tables. For simplicity lets assume we got two rows, one from table A the other from table B. Now the reducer can combine the two rows and output the combined row. This will work just the same if there were multiple rows from each table with the same "a|b" key, in that case the reducer would have to generate the Cartesian product of all the rows. Outer joins can also be done this way, in an outer join you only get one row in the reducer for a given "a|b" key but still generate an output.

-eran

On Fri, Jun 3, 2011 at 00:05, Michael Segel <[EMAIL PROTECTED]>wrote:

>
> Not to beat a dead horse, but I thought a bit more about the problem.
> If you want to do this all in HBase using a M/R job...
>
> Lets define the following:
> SELECT *
> FROM A, B
> WHERE A.a = B.a
> AND     A.b = B.b
> AND     A.c = xxx
> AND     A.d = yyy
> AND     B.e = zzz
>
> Is the sample query.
>
> So our join key is "a|b" because we're matching on columns a and b.
> (The pipe is to delimit the columns, assuming the columns themselves
> don't contain pipes...)
>
> Our filters on A are c and d while e is the filter on B.
>
> So we want to do the following:
>
> M/R Map job 1 gets the subset from table A along with a set of unique keys.
> M/R Map job 2 gets the subset from table B along with a set of unique keys.
> M/R Map job 3 takes either set of unique keys as the input list and
> you split it based on the number of parallel mappers you want to use.

Well....

David, is correct.

Eran wanted to do a join which is a relational concept that isn't natively supported by a NoSQL database. A better model would be a hierarchical model like Dick Pick's Revelation. (Univers aka U2 from Ardent/Informix/IBM/now JRockit?).
And yes, we're looking back 40 some odd years in to either a merge/sort solution or how databases do a relational join. :-)

Eran wants to do this in a single m/r job. The short answer is you can't.  Longer answer is that if your main class implements Tool Runner, you can launch two jobs in parallel to get your subsets, and then when they both complete, you run the join job on them. So I guess its a single 'job' or rather app. :-)

With respect to Doug's posts, you can't do a multi-get off the bat because in the general case you're not fetching based on the row key but a column which is not part of the row key. (It could be a foreign key which would mean that at least one of your table fetches will be off the row key but you can't guarantee it.)

So if you don't want to use temp tables, then you have to put your results in a sorted order, and you still want to get the unique set of the join-keys which means you have to run a reduce job. Then you can use the unique key set and then do the scans. (You can't do a multi-get because you're doing a scan with a start and stop row(s).)

The reason I suggest that if you're going to do a join operation, you want to use temp tables because it makes your life easier and probably faster too.

Bottom line... I guess many data architects are going to need rethink their data models when working on big data. :-)

-Mike

PS. If I get a spare moment, I may code this up...
> From: [EMAIL PROTECTED]
> To: [EMAIL PROTECTED]
> Date: Mon, 6 Jun 2011 17:19:44 -0400
> Subject: RE: How to efficiently join HBase tables?
>
> Re:  " So, you all realize the joins have been talked about in the database community for 40 years?"
>
> Great point.  What's old is new!    :-)
>
> My suggested from earlier in the thread was a variant of nested loops by using multi-get in HTable, which would reduce the number of RPC calls.  So it's a "bulk-select nested loops" of sorts (i.e., as opposed to the 1-by-1 lookup of regular nested loops).
>
>
> -----Original Message-----
> From: Buttler, David [mailto:[EMAIL PROTECTED]]
> Sent: Monday, June 06, 2011 4:30 PM
> To: [EMAIL PROTECTED]
> Subject: RE: How to efficiently join HBase tables?
>
> So, you all realize the joins have been talked about in the database community for 40 years?  There are two main types of joins:
> Nested loops
> Hash table
>
> Mike, in his various emails seems to be trying to re-imagine how to implement both types of joins in HBase (which seems like a reasonable goal). I am not exactly sure what Eran is going for here, but it seems like Eran is glossing over a piece.  If you have two scanners for table A and B, then table B needs to be rescanned for every unique part of the join condition in table A.  There are certain ways of improving the efficiency of that: the two most obvious are pushing the selection criteria down to the scans, and scanning all of the same join values from table B at the same time (which requires that Table B's key is the join, or a secondary structure that stores the join values as the primary order).
>
> Dave
>
> -----Original Message-----
> From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran Kutner
> Sent: Friday, June 03, 2011 12:24 AM
> To: [EMAIL PROTECTED]
> Subject: Re: How to efficiently join HBase tables?
>
> Mike, this more or less what I tried to  describe in my initial post, only you explained it much better.
> The problem is that I want to do all of this in one M/R run, not 3 and without explicit temp tables. If there was only a way to feed both table A and table B into the M/R job then it could be done.
>
> Let's take your query and assumptions, for example.
> So we configure scanner A to return rows where c=xxx and d=yyy We then configure scanner B to return rows where e=zzz Now we feed all those rows to the mapper.
     

Re: " With respect to Doug's posts, you can't do a multi-get off the bat"

That's an assumption, but you're entitled to your opinion.

-----Original Message-----
From: Michael Segel [mailto:[EMAIL PROTECTED]]
Sent: Monday, June 06, 2011 10:08 PM
To: [EMAIL PROTECTED]
Subject: RE: How to efficiently join HBase tables?
Well....

David, is correct.

Eran wanted to do a join which is a relational concept that isn't natively supported by a NoSQL database. A better model would be a hierarchical model like Dick Pick's Revelation. (Univers aka U2 from Ardent/Informix/IBM/now JRockit?).
And yes, we're looking back 40 some odd years in to either a merge/sort solution or how databases do a relational join. :-)

Eran wants to do this in a single m/r job. The short answer is you can't.  Longer answer is that if your main class implements Tool Runner, you can launch two jobs in parallel to get your subsets, and then when they both complete, you run the join job on them. So I guess its a single 'job' or rather app. :-)

With respect to Doug's posts, you can't do a multi-get off the bat because in the general case you're not fetching based on the row key but a column which is not part of the row key. (It could be a foreign key which would mean that at least one of your table fetches will be off the row key but you can't guarantee it.)

So if you don't want to use temp tables, then you have to put your results in a sorted order, and you still want to get the unique set of the join-keys which means you have to run a reduce job. Then you can use the unique key set and then do the scans. (You can't do a multi-get because you're doing a scan with a start and stop row(s).)

The reason I suggest that if you're going to do a join operation, you want to use temp tables because it makes your life easier and probably faster too.

Bottom line... I guess many data architects are going to need rethink their data models when working on big data. :-)

-Mike

PS. If I get a spare moment, I may code this up...
> From: [EMAIL PROTECTED]
> To: [EMAIL PROTECTED]
> Date: Mon, 6 Jun 2011 17:19:44 -0400
> Subject: RE: How to efficiently join HBase tables?
>
> Re:  " So, you all realize the joins have been talked about in the database community for 40 years?"
>
> Great point.  What's old is new!    :-)
>
> My suggested from earlier in the thread was a variant of nested loops by using multi-get in HTable, which would reduce the number of RPC calls.  So it's a "bulk-select nested loops" of sorts (i.e., as opposed to the 1-by-1 lookup of regular nested loops).
>
>
> -----Original Message-----
> From: Buttler, David [mailto:[EMAIL PROTECTED]]
> Sent: Monday, June 06, 2011 4:30 PM
> To: [EMAIL PROTECTED]
> Subject: RE: How to efficiently join HBase tables?
>
> So, you all realize the joins have been talked about in the database community for 40 years?  There are two main types of joins:
> Nested loops
> Hash table
>
> Mike, in his various emails seems to be trying to re-imagine how to implement both types of joins in HBase (which seems like a reasonable goal). I am not exactly sure what Eran is going for here, but it seems like Eran is glossing over a piece.  If you have two scanners for table A and B, then table B needs to be rescanned for every unique part of the join condition in table A.  There are certain ways of improving the efficiency of that: the two most obvious are pushing the selection criteria down to the scans, and scanning all of the same join values from table B at the same time (which requires that Table B's key is the join, or a secondary structure that stores the join values as the primary order).
>
> Dave
>
> -----Original Message-----
> From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran
> Kutner
> Sent: Friday, June 03, 2011 12:24 AM
> To: [EMAIL PROTECTED]
> Subject: Re: How to efficiently join HBase tables?
>
> Mike, this more or less what I tried to  describe in my initial post, only you explained it much better.

Unless I am mistaken... get() requires a row key, right?
And you can join tables on column data which isn't in the row key, right?

So how do you do a get()? :-)

Sure there is more than one way to skin a cat. But if you want to be efficient... You will create a set of unique keys based on the columns that you want to join. Note that if you are going to use a temp table in hbase, you will want to store the unique key value A|B and when you write the row to the temp table, you will append an unique identifier like a uuid so that you don't lose the row.

Here your input list to the actual join is going to be the list of unique keys and then you do a scan to get the rows.

Again, I could be wrong but how can you perform a get() when you only know a portion of the row key?

Sent from a remote device. Please excuse any typos...

Mike Segel

On Jun 8, 2011, at 8:01 AM, Doug Meil <[EMAIL PROTECTED]> wrote:

>
> Re: " With respect to Doug's posts, you can't do a multi-get off the bat"
>
> That's an assumption, but you're entitled to your opinion.
>
> -----Original Message-----
> From: Michael Segel [mailto:[EMAIL PROTECTED]]
> Sent: Monday, June 06, 2011 10:08 PM
> To: [EMAIL PROTECTED]
> Subject: RE: How to efficiently join HBase tables?
>
>
> Well....
>
> David, is correct.
>
> Eran wanted to do a join which is a relational concept that isn't natively supported by a NoSQL database. A better model would be a hierarchical model like Dick Pick's Revelation. (Univers aka U2 from Ardent/Informix/IBM/now JRockit?).
> And yes, we're looking back 40 some odd years in to either a merge/sort solution or how databases do a relational join. :-)
>
> Eran wants to do this in a single m/r job. The short answer is you can't.  Longer answer is that if your main class implements Tool Runner, you can launch two jobs in parallel to get your subsets, and then when they both complete, you run the join job on them. So I guess its a single 'job' or rather app. :-)
>
> With respect to Doug's posts, you can't do a multi-get off the bat because in the general case you're not fetching based on the row key but a column which is not part of the row key. (It could be a foreign key which would mean that at least one of your table fetches will be off the row key but you can't guarantee it.)
>
> So if you don't want to use temp tables, then you have to put your results in a sorted order, and you still want to get the unique set of the join-keys which means you have to run a reduce job. Then you can use the unique key set and then do the scans. (You can't do a multi-get because you're doing a scan with a start and stop row(s).)
>
> The reason I suggest that if you're going to do a join operation, you want to use temp tables because it makes your life easier and probably faster too.
>
> Bottom line... I guess many data architects are going to need rethink their data models when working on big data. :-)
>
> -Mike
>
> PS. If I get a spare moment, I may code this up...
>
>
>> From: [EMAIL PROTECTED]
>> To: [EMAIL PROTECTED]
>> Date: Mon, 6 Jun 2011 17:19:44 -0400
>> Subject: RE: How to efficiently join HBase tables?
>>
>> Re:  " So, you all realize the joins have been talked about in the database community for 40 years?"
>>
>> Great point.  What's old is new!    :-)
>>
>> My suggested from earlier in the thread was a variant of nested loops by using multi-get in HTable, which would reduce the number of RPC calls.  So it's a "bulk-select nested loops" of sorts (i.e., as opposed to the 1-by-1 lookup of regular nested loops).
>>
>>
>> -----Original Message-----
>> From: Buttler, David [mailto:[EMAIL PROTECTED]]
>> Sent: Monday, June 06, 2011 4:30 PM
>> To: [EMAIL PROTECTED]
>> Subject: RE: How to efficiently join HBase tables?
>>
>> So, you all realize the joins have been talked about in the database community for 40 years?  There are two main types of joins:
>> Nested loops
>> Hash table
>>
>> Mike, in his various emails seems to be trying to re-imagine how to implement both types of joins in HBase (which seems like a reasonable goal). I am not exactly sure what Eran is going for here, but it seems like Eran is glossing over a piece.  If you have two scanners for table A and B, then table B needs to be rescanned for every unique part of the join condition in table A.  There are certain ways of improving the efficiency of that: the two most obvious are pushing the selection criteria down to the scans, and scanning all of the same join values from table B at the same time (which requires that Table B's key is the join, or a secondary structure that stores the join values as the primary order).

I'd like to clarify, again what I'm trying to do and why I still think it's
the best way to do it.
I want to join two large tables, I'm assuming, and this is the key to the
efficiency of this method, that: 1) I'm getting a lot of data from table A,
something which is close enough top a full table scan, and 2) this implies
that I will need to join with most of table B as well.
All the suggestions from the SQL world are doing lookups, one way or another
in table B. My suggestion is to use the power of the shuffle phase to do the
join. It is obviously doable, so I don't understand the statement that it
can't be done.
So to go over it again:
1. You feed all the rows from table A and B into the mapper.
2. For each row, the mapper should output a new row with a key constructed
from the join fields and a value which is the row itself (same as the input
value it got).
3. The shuffle phase will make sure all rows with the same values in the
join fields will end up together.
4. The reducer will get all the rows for a single set of join field values
together and perform the actual join. The reducer can be programmed to do an
inner or outer join at this point.

I can't prove it without actually writing and testing it but I have a strong
feeling this will be much more efficient for large joins than any form of
lookup.

-eran

On Wed, Jun 8, 2011 at 16:01, Doug Meil <[EMAIL PROTECTED]>wrote:

>
> Re: " With respect to Doug's posts, you can't do a multi-get off the bat"
>
> That's an assumption, but you're entitled to your opinion.
>
> -----Original Message-----
> From: Michael Segel [mailto:[EMAIL PROTECTED]]
> Sent: Monday, June 06, 2011 10:08 PM
> To: [EMAIL PROTECTED]
> Subject: RE: How to efficiently join HBase tables?
>
>
> Well....
>
> David, is correct.
>
> Eran wanted to do a join which is a relational concept that isn't natively
> supported by a NoSQL database. A better model would be a hierarchical model
> like Dick Pick's Revelation. (Univers aka U2 from Ardent/Informix/IBM/now
> JRockit?).
> And yes, we're looking back 40 some odd years in to either a merge/sort
> solution or how databases do a relational join. :-)
>
> Eran wants to do this in a single m/r job. The short answer is you can't.
>  Longer answer is that if your main class implements Tool Runner, you can
> launch two jobs in parallel to get your subsets, and then when they both
> complete, you run the join job on them. So I guess its a single 'job' or
> rather app. :-)
>
> With respect to Doug's posts, you can't do a multi-get off the bat because
> in the general case you're not fetching based on the row key but a column
> which is not part of the row key. (It could be a foreign key which would
> mean that at least one of your table fetches will be off the row key but you
> can't guarantee it.)
>
> So if you don't want to use temp tables, then you have to put your results
> in a sorted order, and you still want to get the unique set of the join-keys
> which means you have to run a reduce job. Then you can use the unique key
> set and then do the scans. (You can't do a multi-get because you're doing a
> scan with a start and stop row(s).)
>
> The reason I suggest that if you're going to do a join operation, you want
> to use temp tables because it makes your life easier and probably faster
> too.
>
> Bottom line... I guess many data architects are going to need rethink their
> data models when working on big data. :-)
>
> -Mike
>
> PS. If I get a spare moment, I may code this up...
>
>
> > From: [EMAIL PROTECTED]
> > To: [EMAIL PROTECTED]
> > Date: Mon, 6 Jun 2011 17:19:44 -0400
> > Subject: RE: How to efficiently join HBase tables?
> >
> > Re:  " So, you all realize the joins have been talked about in the
> database community for 40 years?"
> >
> > Great point.  What's old is new!    :-)
> >
> > My suggested from earlier in the thread was a variant of nested loops by
> using multi-get in HTable, which would reduce the number of RPC calls.  So

Let's make a toy example to see if we can capture all of the edge conditions:
Table A
-------
Key1 joinVal_1
Key2 joinVal_2
Key3 joinVal_1

Table B
-------
Key4 joinVal_1
Key5 joinVal_3
Key6 joinVal_2

Now, assume that we have a mapper that takes two values, one row from A, and one row from B.  Are you suggesting that we get the following map calls:
Key1 & key4
Key2 & key5
Key3 & key6

Or are you suggesting we get the following:
Key1 & key4
Key1 & key5
Key1 & key6
Key2 & key4
Key2 & key5
Key2 & key6
Key3 & key4
Key3 & key5
Key3 & key6

Or are you suggesting something different?

Dave

-----Original Message-----
From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran Kutner
Sent: Wednesday, June 08, 2011 11:47 AM
To: [EMAIL PROTECTED]
Subject: Re: How to efficiently join HBase tables?

I'd like to clarify, again what I'm trying to do and why I still think it's
the best way to do it.
I want to join two large tables, I'm assuming, and this is the key to the
efficiency of this method, that: 1) I'm getting a lot of data from table A,
something which is close enough top a full table scan, and 2) this implies
that I will need to join with most of table B as well.
All the suggestions from the SQL world are doing lookups, one way or another
in table B. My suggestion is to use the power of the shuffle phase to do the
join. It is obviously doable, so I don't understand the statement that it
can't be done.
So to go over it again:
1. You feed all the rows from table A and B into the mapper.
2. For each row, the mapper should output a new row with a key constructed
from the join fields and a value which is the row itself (same as the input
value it got).
3. The shuffle phase will make sure all rows with the same values in the
join fields will end up together.
4. The reducer will get all the rows for a single set of join field values
together and perform the actual join. The reducer can be programmed to do an
inner or outer join at this point.

I can't prove it without actually writing and testing it but I have a strong
feeling this will be much more efficient for large joins than any form of
lookup.

-eran

On Wed, Jun 8, 2011 at 16:01, Doug Meil <[EMAIL PROTECTED]>wrote:

>
> Re: " With respect to Doug's posts, you can't do a multi-get off the bat"
>
> That's an assumption, but you're entitled to your opinion.
>
> -----Original Message-----
> From: Michael Segel [mailto:[EMAIL PROTECTED]]
> Sent: Monday, June 06, 2011 10:08 PM
> To: [EMAIL PROTECTED]
> Subject: RE: How to efficiently join HBase tables?
>
>
> Well....
>
> David, is correct.
>
> Eran wanted to do a join which is a relational concept that isn't natively
> supported by a NoSQL database. A better model would be a hierarchical model
> like Dick Pick's Revelation. (Univers aka U2 from Ardent/Informix/IBM/now
> JRockit?).
> And yes, we're looking back 40 some odd years in to either a merge/sort
> solution or how databases do a relational join. :-)
>
> Eran wants to do this in a single m/r job. The short answer is you can't.
>  Longer answer is that if your main class implements Tool Runner, you can
> launch two jobs in parallel to get your subsets, and then when they both
> complete, you run the join job on them. So I guess its a single 'job' or
> rather app. :-)
>
> With respect to Doug's posts, you can't do a multi-get off the bat because
> in the general case you're not fetching based on the row key but a column
> which is not part of the row key. (It could be a foreign key which would
> mean that at least one of your table fetches will be off the row key but you
> can't guarantee it.)
>
> So if you don't want to use temp tables, then you have to put your results
> in a sorted order, and you still want to get the unique set of the join-keys
> which means you have to run a reduce job. Then you can use the unique key
> set and then do the scans. (You can't do a multi-get because you're doing a
> scan with a start and stop row(s).)
>
> The reason I suggest that if you're going to do a join operation, you want

I believe this is what Eran is suggesting:

Table A
-------
Row1 (has joinVal_1)
Row2 (has joinVal_2)
Row3 (has joinVal_1)

Table B
-------
Row4 (has joinVal_1)
Row5 (has joinVal_3)
Row6 (has joinVal_2)

Mapper receives a list of input rows (union of both input tables in any
order), and produces (=>) intermediate key, value pairs, where the key is
the join field, and the value is whatever portion of the row you want
available in your output

Map
----------
A, Row1 => (joinVal_1, [A,Row1])
A, Row2 => (joinVal_2, [A,Row2])
A, Row3 => (joinVal_1, [A,Row3])
B, Row4 => (joinVal_1, [B,Row4])
B, Row5 => (joinVal_3, [B,Row5])
B, Row6 => (joinVal_2, [B,Row6])

Shuffle phase partitions and sorts by the map output key (which is the join
value)
The Reduce phase then gets a key for the join value and a list of values
containing all of the input rows (from either table) with that join value.
It can then perform whatever operations you want (like enumerate the subset
of the Cartesian product for that join value)

Reduce
------------
joinVal_1, {[A,Row1], [A,Row3], [B,Row4]} => Row1 x Row4, Row3 x Row4
joinVal_2, {[A,Row2], [B,Row6]} => Row2 x Row6
joinVal_3, {[A,Row3]} => {}

This sounds effective to me, so long as you can perform any desired
operations on the all the rows matching a single join value via a single
iteration through the stream of reduce input values (for example, if the set
of data for each join value fits in memory).  Otherwise you'd need to put
the list of matches from table A some place that you can iterate over it
again for each match in table B.

Dave

On Wed, Jun 8, 2011 at 1:45 PM, Buttler, David <[EMAIL PROTECTED]> wrote:

> Let's make a toy example to see if we can capture all of the edge
> conditions:
> Table A
> -------
> Key1 joinVal_1
> Key2 joinVal_2
> Key3 joinVal_1
>
> Table B
> -------
> Key4 joinVal_1
> Key5 joinVal_3
> Key6 joinVal_2
>
> Now, assume that we have a mapper that takes two values, one row from A,
> and one row from B.  Are you suggesting that we get the following map calls:
> Key1 & key4
> Key2 & key5
> Key3 & key6
>
> Or are you suggesting we get the following:
> Key1 & key4
> Key1 & key5
> Key1 & key6
> Key2 & key4
> Key2 & key5
> Key2 & key6
> Key3 & key4
> Key3 & key5
> Key3 & key6
>
> Or are you suggesting something different?
>
> Dave
>
> -----Original Message-----
> From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran
> Kutner
> Sent: Wednesday, June 08, 2011 11:47 AM
> To: [EMAIL PROTECTED]
> Subject: Re: How to efficiently join HBase tables?
>
> I'd like to clarify, again what I'm trying to do and why I still think it's
> the best way to do it.
> I want to join two large tables, I'm assuming, and this is the key to the
> efficiency of this method, that: 1) I'm getting a lot of data from table A,
> something which is close enough top a full table scan, and 2) this implies
> that I will need to join with most of table B as well.
> All the suggestions from the SQL world are doing lookups, one way or
> another
> in table B. My suggestion is to use the power of the shuffle phase to do
> the
> join. It is obviously doable, so I don't understand the statement that it
> can't be done.
> So to go over it again:
> 1. You feed all the rows from table A and B into the mapper.
> 2. For each row, the mapper should output a new row with a key constructed
> from the join fields and a value which is the row itself (same as the input
> value it got).
> 3. The shuffle phase will make sure all rows with the same values in the
> join fields will end up together.
> 4. The reducer will get all the rows for a single set of join field values
> together and perform the actual join. The reducer can be programmed to do
> an
> inner or outer join at this point.
>
> I can't prove it without actually writing and testing it but I have a
> strong
> feeling this will be much more efficient for large joins than any form of
> lookup.
>
> -eran
>
>
>
> On Wed, Jun 8, 2011 at 16:01, Doug Meil <[EMAIL PROTECTED]

Thank you for the explanation, I think I understand the suggestion now.  I completely agree with you that this would be effective for cases that you can do the join of the sorted values in memory.

A small tweak would make this more generic and effective for any size. If you had two separate Map/Reduce jobs for tables A and B, you could simple perform a merge join on the outputs that is very efficient (only touches data items once in each of the A and B output).  This, obviously, makes it larger than a single M/R job, but I don't think that is a major drawback.

Dave
-----Original Message-----
From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Dave Latham
Sent: Wednesday, June 08, 2011 2:36 PM
To: [EMAIL PROTECTED]
Subject: Re: How to efficiently join HBase tables?

I believe this is what Eran is suggesting:

Table A
-------
Row1 (has joinVal_1)
Row2 (has joinVal_2)
Row3 (has joinVal_1)

Table B
-------
Row4 (has joinVal_1)
Row5 (has joinVal_3)
Row6 (has joinVal_2)

Mapper receives a list of input rows (union of both input tables in any
order), and produces (=>) intermediate key, value pairs, where the key is
the join field, and the value is whatever portion of the row you want
available in your output

Map
----------
A, Row1 => (joinVal_1, [A,Row1])
A, Row2 => (joinVal_2, [A,Row2])
A, Row3 => (joinVal_1, [A,Row3])
B, Row4 => (joinVal_1, [B,Row4])
B, Row5 => (joinVal_3, [B,Row5])
B, Row6 => (joinVal_2, [B,Row6])

Shuffle phase partitions and sorts by the map output key (which is the join
value)
The Reduce phase then gets a key for the join value and a list of values
containing all of the input rows (from either table) with that join value.
It can then perform whatever operations you want (like enumerate the subset
of the Cartesian product for that join value)

Reduce
------------
joinVal_1, {[A,Row1], [A,Row3], [B,Row4]} => Row1 x Row4, Row3 x Row4
joinVal_2, {[A,Row2], [B,Row6]} => Row2 x Row6
joinVal_3, {[A,Row3]} => {}

This sounds effective to me, so long as you can perform any desired
operations on the all the rows matching a single join value via a single
iteration through the stream of reduce input values (for example, if the set
of data for each join value fits in memory).  Otherwise you'd need to put
the list of matches from table A some place that you can iterate over it
again for each match in table B.

Dave

On Wed, Jun 8, 2011 at 1:45 PM, Buttler, David <[EMAIL PROTECTED]> wrote:

> Let's make a toy example to see if we can capture all of the edge
> conditions:
> Table A
> -------
> Key1 joinVal_1
> Key2 joinVal_2
> Key3 joinVal_1
>
> Table B
> -------
> Key4 joinVal_1
> Key5 joinVal_3
> Key6 joinVal_2
>
> Now, assume that we have a mapper that takes two values, one row from A,
> and one row from B.  Are you suggesting that we get the following map calls:
> Key1 & key4
> Key2 & key5
> Key3 & key6
>
> Or are you suggesting we get the following:
> Key1 & key4
> Key1 & key5
> Key1 & key6
> Key2 & key4
> Key2 & key5
> Key2 & key6
> Key3 & key4
> Key3 & key5
> Key3 & key6
>
> Or are you suggesting something different?
>
> Dave
>
> -----Original Message-----
> From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran
> Kutner
> Sent: Wednesday, June 08, 2011 11:47 AM
> To: [EMAIL PROTECTED]
> Subject: Re: How to efficiently join HBase tables?
>
> I'd like to clarify, again what I'm trying to do and why I still think it's
> the best way to do it.
> I want to join two large tables, I'm assuming, and this is the key to the
> efficiency of this method, that: 1) I'm getting a lot of data from table A,
> something which is close enough top a full table scan, and 2) this implies
> that I will need to join with most of table B as well.
> All the suggestions from the SQL world are doing lookups, one way or
> another
> in table B. My suggestion is to use the power of the shuffle phase to do
> the
> join. It is obviously doable, so I don't understand the statement that it
> can't be done.
> So to go over it again:

Hi there-

Summary comment:

1)  Preference

Several people in this thread have suggested approaches (map-side memory join, multi-get, temp files), all of which have merit and have advantages in certain situations.  Kudos to the dist-list for chiming in.  The "right" approach depends on the specific problem you are trying to solve, and that's preference.

2)   Possibility

Mike, you're the only one in this thread arguing that one of the approaches isn't possible.  And you seem to be arguing that it's never possible to look up a record in HBase with a Get, but others don't seem to have this problem.

I wish everybody success with their HBase join research, but I'm checking out of this thread (ka-ching).
-----Original Message-----
From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Michel Segel
Sent: Wednesday, June 08, 2011 10:14 AM
To: [EMAIL PROTECTED]
Subject: Re: How to efficiently join HBase tables?

Unless I am mistaken... get() requires a row key, right?
And you can join tables on column data which isn't in the row key, right?

So how do you do a get()? :-)

Sure there is more than one way to skin a cat. But if you want to be efficient... You will create a set of unique keys based on the columns that you want to join. Note that if you are going to use a temp table in hbase, you will want to store the unique key value A|B and when you write the row to the temp table, you will append an unique identifier like a uuid so that you don't lose the row.

Here your input list to the actual join is going to be the list of unique keys and then you do a scan to get the rows.

Again, I could be wrong but how can you perform a get() when you only know a portion of the row key?

Sent from a remote device. Please excuse any typos...

Mike Segel

On Jun 8, 2011, at 8:01 AM, Doug Meil <[EMAIL PROTECTED]> wrote:

>
> Re: " With respect to Doug's posts, you can't do a multi-get off the bat"
>
> That's an assumption, but you're entitled to your opinion.
>
> -----Original Message-----
> From: Michael Segel [mailto:[EMAIL PROTECTED]]
> Sent: Monday, June 06, 2011 10:08 PM
> To: [EMAIL PROTECTED]
> Subject: RE: How to efficiently join HBase tables?
>
>
> Well....
>
> David, is correct.
>
> Eran wanted to do a join which is a relational concept that isn't natively supported by a NoSQL database. A better model would be a hierarchical model like Dick Pick's Revelation. (Univers aka U2 from Ardent/Informix/IBM/now JRockit?).
> And yes, we're looking back 40 some odd years in to either a
> merge/sort solution or how databases do a relational join. :-)
>
> Eran wants to do this in a single m/r job. The short answer is you
> can't.  Longer answer is that if your main class implements Tool
> Runner, you can launch two jobs in parallel to get your subsets, and
> then when they both complete, you run the join job on them. So I guess
> its a single 'job' or rather app. :-)
>
> With respect to Doug's posts, you can't do a multi-get off the bat
> because in the general case you're not fetching based on the row key
> but a column which is not part of the row key. (It could be a foreign
> key which would mean that at least one of your table fetches will be
> off the row key but you can't guarantee it.)
>
> So if you don't want to use temp tables, then you have to put your
> results in a sorted order, and you still want to get the unique set of
> the join-keys which means you have to run a reduce job. Then you can
> use the unique key set and then do the scans. (You can't do a
> multi-get because you're doing a scan with a start and stop row(s).)
>
> The reason I suggest that if you're going to do a join operation, you want to use temp tables because it makes your life easier and probably faster too.
>
> Bottom line... I guess many data architects are going to need rethink
> their data models when working on big data. :-)
>
> -Mike
>
> PS. If I get a spare moment, I may code this up...
>
>
>> From: [EMAIL PROTECTED]

Exactly!
Thanks Dave for a much better explanation than mine!

-eran

On Thu, Jun 9, 2011 at 00:35, Dave Latham <[EMAIL PROTECTED]> wrote:

> I believe this is what Eran is suggesting:
>
> Table A
> -------
> Row1 (has joinVal_1)
> Row2 (has joinVal_2)
> Row3 (has joinVal_1)
>
> Table B
> -------
> Row4 (has joinVal_1)
> Row5 (has joinVal_3)
> Row6 (has joinVal_2)
>
> Mapper receives a list of input rows (union of both input tables in any
> order), and produces (=>) intermediate key, value pairs, where the key is
> the join field, and the value is whatever portion of the row you want
> available in your output
>
> Map
> ----------
> A, Row1 => (joinVal_1, [A,Row1])
> A, Row2 => (joinVal_2, [A,Row2])
> A, Row3 => (joinVal_1, [A,Row3])
> B, Row4 => (joinVal_1, [B,Row4])
> B, Row5 => (joinVal_3, [B,Row5])
> B, Row6 => (joinVal_2, [B,Row6])
>
> Shuffle phase partitions and sorts by the map output key (which is the join
> value)
> The Reduce phase then gets a key for the join value and a list of values
> containing all of the input rows (from either table) with that join value.
> It can then perform whatever operations you want (like enumerate the subset
> of the Cartesian product for that join value)
>
> Reduce
> ------------
> joinVal_1, {[A,Row1], [A,Row3], [B,Row4]} => Row1 x Row4, Row3 x Row4
> joinVal_2, {[A,Row2], [B,Row6]} => Row2 x Row6
> joinVal_3, {[A,Row3]} => {}
>
> This sounds effective to me, so long as you can perform any desired
> operations on the all the rows matching a single join value via a single
> iteration through the stream of reduce input values (for example, if the
> set
> of data for each join value fits in memory).  Otherwise you'd need to put
> the list of matches from table A some place that you can iterate over it
> again for each match in table B.
>
> Dave
>
> On Wed, Jun 8, 2011 at 1:45 PM, Buttler, David <[EMAIL PROTECTED]> wrote:
>
> > Let's make a toy example to see if we can capture all of the edge
> > conditions:
> > Table A
> > -------
> > Key1 joinVal_1
> > Key2 joinVal_2
> > Key3 joinVal_1
> >
> > Table B
> > -------
> > Key4 joinVal_1
> > Key5 joinVal_3
> > Key6 joinVal_2
> >
> > Now, assume that we have a mapper that takes two values, one row from A,
> > and one row from B.  Are you suggesting that we get the following map
> calls:
> > Key1 & key4
> > Key2 & key5
> > Key3 & key6
> >
> > Or are you suggesting we get the following:
> > Key1 & key4
> > Key1 & key5
> > Key1 & key6
> > Key2 & key4
> > Key2 & key5
> > Key2 & key6
> > Key3 & key4
> > Key3 & key5
> > Key3 & key6
> >
> > Or are you suggesting something different?
> >
> > Dave
> >
> > -----Original Message-----
> > From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran
> > Kutner
> > Sent: Wednesday, June 08, 2011 11:47 AM
> > To: [EMAIL PROTECTED]
> > Subject: Re: How to efficiently join HBase tables?
> >
> > I'd like to clarify, again what I'm trying to do and why I still think
> it's
> > the best way to do it.
> > I want to join two large tables, I'm assuming, and this is the key to the
> > efficiency of this method, that: 1) I'm getting a lot of data from table
> A,
> > something which is close enough top a full table scan, and 2) this
> implies
> > that I will need to join with most of table B as well.
> > All the suggestions from the SQL world are doing lookups, one way or
> > another
> > in table B. My suggestion is to use the power of the shuffle phase to do
> > the
> > join. It is obviously doable, so I don't understand the statement that it
> > can't be done.
> > So to go over it again:
> > 1. You feed all the rows from table A and B into the mapper.
> > 2. For each row, the mapper should output a new row with a key
> constructed
> > from the join fields and a value which is the row itself (same as the
> input
> > value it got).
> > 3. The shuffle phase will make sure all rows with the same values in the
> > join fields will end up together.
> > 4. The reducer will get all the rows for a single set of join field

Doug,
I think I should clarify something...

Yes I am the only one who is saying get() won't work.
The question was asked on how to do an efficient join where there were no specific parameters like joining on key values. It wasn't until yesterday that Eran gave an example of the specific problem...

So you have to make some assumptions that you are attempting to solve this problem in the general case. That is you cant assume that you are joining on the row keys. Since we are talking about a big data problem you can assume that your data sets are going to be huge.
So you have to consider how many rows you can store in memory and that you may not have enough memory.

Because most who are moving to a NoSQL database only have been exposed to relational models, they will approach HBase from a relational schema design. So solving the question of how to join two tables efficiently in general terms has a lot of value to the community as a whole.

David is right in that I'm looking back and taking the approach of how joining two data sets.
It's not rocket science and this problem has been solved under different paradigms.

Sent from a remote device. Please excuse any typos...

Mike Segel

On Jun 8, 2011, at 9:56 PM, Doug Meil <[EMAIL PROTECTED]> wrote:

> Hi there-
>
> Summary comment:
>
> 1)  Preference
>
> Several people in this thread have suggested approaches (map-side memory join, multi-get, temp files), all of which have merit and have advantages in certain situations.  Kudos to the dist-list for chiming in.  The "right" approach depends on the specific problem you are trying to solve, and that's preference.
>
> 2)   Possibility
>
> Mike, you're the only one in this thread arguing that one of the approaches isn't possible.  And you seem to be arguing that it's never possible to look up a record in HBase with a Get, but others don't seem to have this problem.
>
> I wish everybody success with their HBase join research, but I'm checking out of this thread (ka-ching).
>
>
> -----Original Message-----
> From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Michel Segel
> Sent: Wednesday, June 08, 2011 10:14 AM
> To: [EMAIL PROTECTED]
> Subject: Re: How to efficiently join HBase tables?
>
> Unless I am mistaken... get() requires a row key, right?
> And you can join tables on column data which isn't in the row key, right?
>
> So how do you do a get()? :-)
>
> Sure there is more than one way to skin a cat. But if you want to be efficient... You will create a set of unique keys based on the columns that you want to join. Note that if you are going to use a temp table in hbase, you will want to store the unique key value A|B and when you write the row to the temp table, you will append an unique identifier like a uuid so that you don't lose the row.
>
> Here your input list to the actual join is going to be the list of unique keys and then you do a scan to get the rows.
>
> Again, I could be wrong but how can you perform a get() when you only know a portion of the row key?
>
>
>
> Sent from a remote device. Please excuse any typos...
>
> Mike Segel
>
> On Jun 8, 2011, at 8:01 AM, Doug Meil <[EMAIL PROTECTED]> wrote:
>
>>
>> Re: " With respect to Doug's posts, you can't do a multi-get off the bat"
>>
>> That's an assumption, but you're entitled to your opinion.
>>
>> -----Original Message-----
>> From: Michael Segel [mailto:[EMAIL PROTECTED]]
>> Sent: Monday, June 06, 2011 10:08 PM
>> To: [EMAIL PROTECTED]
>> Subject: RE: How to efficiently join HBase tables?
>>
>>
>> Well....
>>
>> David, is correct.
>>
>> Eran wanted to do a join which is a relational concept that isn't natively supported by a NoSQL database. A better model would be a hierarchical model like Dick Pick's Revelation. (Univers aka U2 from Ardent/Informix/IBM/now JRockit?).
>> And yes, we're looking back 40 some odd years in to either a
>> merge/sort solution or how databases do a relational join. :-)

>> This sounds effective to me, so long as you can perform any desired
>> operations on the all the rows matching a single join value via a single
>> iteration through the stream of reduce input values (for example, if the
>> set
>> of data for each join value fits in memory).  Otherwise you'd need to put
>> the list of matches from table A some place that you can iterate over it
>> again for each match in table B.
This is why I was suggesting using the temp tables and not trying to do it as a sing map reduce job.  When your data sets get very large you will have problems.........

;-)

Sent from a remote device. Please excuse any typos...

Mike Segel

On Jun 9, 2011, at 4:35 AM, Eran Kutner <[EMAIL PROTECTED]> wrote:

> Exactly!
> Thanks Dave for a much better explanation than mine!
>
> -eran
>
>
>
> On Thu, Jun 9, 2011 at 00:35, Dave Latham <[EMAIL PROTECTED]> wrote:
>
>> I believe this is what Eran is suggesting:
>>
>> Table A
>> -------
>> Row1 (has joinVal_1)
>> Row2 (has joinVal_2)
>> Row3 (has joinVal_1)
>>
>> Table B
>> -------
>> Row4 (has joinVal_1)
>> Row5 (has joinVal_3)
>> Row6 (has joinVal_2)
>>
>> Mapper receives a list of input rows (union of both input tables in any
>> order), and produces (=>) intermediate key, value pairs, where the key is
>> the join field, and the value is whatever portion of the row you want
>> available in your output
>>
>> Map
>> ----------
>> A, Row1 => (joinVal_1, [A,Row1])
>> A, Row2 => (joinVal_2, [A,Row2])
>> A, Row3 => (joinVal_1, [A,Row3])
>> B, Row4 => (joinVal_1, [B,Row4])
>> B, Row5 => (joinVal_3, [B,Row5])
>> B, Row6 => (joinVal_2, [B,Row6])
>>
>> Shuffle phase partitions and sorts by the map output key (which is the join
>> value)
>> The Reduce phase then gets a key for the join value and a list of values
>> containing all of the input rows (from either table) with that join value.
>> It can then perform whatever operations you want (like enumerate the subset
>> of the Cartesian product for that join value)
>>
>> Reduce
>> ------------
>> joinVal_1, {[A,Row1], [A,Row3], [B,Row4]} => Row1 x Row4, Row3 x Row4
>> joinVal_2, {[A,Row2], [B,Row6]} => Row2 x Row6
>> joinVal_3, {[A,Row3]} => {}
>>
>>>> This sounds effective to me, so long as you can perform any desired
>>>> operations on the all the rows matching a single join value via a single
>>>> iteration through the stream of reduce input values (for example, if the
>>>> set
>>>> of data for each join value fits in memory).  Otherwise you'd need to put
>>>> the list of matches from table A some place that you can iterate over it
>>>> again for each match in table B.
>>
>> Dave
>>
>> On Wed, Jun 8, 2011 at 1:45 PM, Buttler, David <[EMAIL PROTECTED]> wrote:
>>
>>> Let's make a toy example to see if we can capture all of the edge
>>> conditions:
>>> Table A
>>> -------
>>> Key1 joinVal_1
>>> Key2 joinVal_2
>>> Key3 joinVal_1
>>>
>>> Table B
>>> -------
>>> Key4 joinVal_1
>>> Key5 joinVal_3
>>> Key6 joinVal_2
>>>
>>> Now, assume that we have a mapper that takes two values, one row from A,
>>> and one row from B.  Are you suggesting that we get the following map
>> calls:
>>> Key1 & key4
>>> Key2 & key5
>>> Key3 & key6
>>>
>>> Or are you suggesting we get the following:
>>> Key1 & key4
>>> Key1 & key5
>>> Key1 & key6
>>> Key2 & key4
>>> Key2 & key5
>>> Key2 & key6
>>> Key3 & key4
>>> Key3 & key5
>>> Key3 & key6
>>>
>>> Or are you suggesting something different?
>>>
>>> Dave
>>>
>>> -----Original Message-----
>>> From: [EMAIL PROTECTED] [mailto:[EMAIL PROTECTED]] On Behalf Of Eran
>>> Kutner
>>> Sent: Wednesday, June 08, 2011 11:47 AM
>>> To: [EMAIL PROTECTED]
>>> Subject: Re: How to efficiently join HBase tables?
>>>
>>> I'd like to clarify, again what I'm trying to do and why I still think
>> it's
>>> the best way to do it.
>>> I want to join two large tables, I'm assuming, and this is the key to the
>>> efficiency of this method, that: 1) I'm getting a lot of data from table
>> A,
>>> something which is close enough top a full table scan, and 2) this

Hello!
   Regarding the same subject of joining, I have the following scenario:
1. I have a big table DOCS that contains the columns
      UUID DOCID
      sdsd  1
      hdhs  3
      gdhg  7
      shdg  9
  
and so on (hope you got the idea)
2. an external list of docID
(LIST)
   3
   1
   7

 upon a I have to query("join") the DOCS DOCID column, so that the result should
be   hdhs, sdsd, gdhg. How I can implement such a request? Can be this a
possible solution:
1. to add a new column LIST (in the same column family ) to the DOCS
2 add a new record in it that contain my LIST of docID
3. "Join" column LIST with DOCID column? ( perhaps a weird idea)

Thank you.
 Regards,
 Florin

Depends on a couple of things.  If your LIST is a permanent feature of your document, then it might make sense to add the list(Boolean? Or the list index if the list has a particular sort order) to the doc record.  Otherwise, a little simple programming can get you the results you want:
1) Sort the list (if it is big, then a map reduce job with an identity map / single identity reducer would do the job).  If you require the order of the list to be maintained then you need to add another field to the list indicating order, so that you can recover that after the join.
2) output a list of DOCID / UUID sorted on DOCID
3) use a double iterator through your two outputs to find the UUIDs from the list (and optionally its order in the list)
4) optionally resort the UUID list by the list order index

This will not be particularly fast, but it should be robust to large list sizes.

If your list can fit into the memory of a map task, then put it in a hash map for each Map job, and while you iterate over your docs table, you can only output UUIDs and sort order, and let your reducer reorder them according to your list order.

Dave

-----Original Message-----
From: Florin P [mailto:[EMAIL PROTECTED]]
Sent: Thursday, June 16, 2011 5:44 AM
To: [EMAIL PROTECTED]
Subject: Re: How to efficiently join HBase tables?

 Hello!
   Regarding the same subject of joining, I have the following scenario:
1. I have a big table DOCS that contains the columns
      UUID DOCID
      sdsd  1
      hdhs  3
      gdhg  7
      shdg  9
  
and so on (hope you got the idea)
2. an external list of docID
(LIST)
   3
   1
   7

 upon a I have to query("join") the DOCS DOCID column, so that the result should
be   hdhs, sdsd, gdhg. How I can implement such a request? Can be this a
possible solution:
1. to add a new column LIST (in the same column family ) to the DOCS
2 add a new record in it that contain my LIST of docID
3. "Join" column LIST with DOCID column? ( perhaps a weird idea)

Thank you.
 Regards,
 Florin