You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@hbase.apache.org by Jim Twensky <ji...@gmail.com> on 2008/12/22 21:37:39 UTC
Using Hbase as data sink
Hello,
I have an application which is similar to the word count example given on
the Hadoop Map/Reduce tutorial. Instead of counting the words however, I
count the phrases. That is, for a sentence like:
"How are you"
I emit the following phrases inside my mapper:
How 1
How are 1
How are you 1
are 1
are you 1
you 1
and then inside the reducer, I aggregate the same keys and send them to an
output file.
However, I want to load these (phrase,count) pairs to Hbase instead of
storing them in a file. I've already written the code and it works but I
have some concerns about its performance and I'm not sure if this is the
right way to do it. Here is how my reducer looks:
public class Reduce extends MapReduceBase implements Reducer<Text,
IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterator<IntWritable> values,
OutputCollector<Text, IntWritable> output, Reporter reporter) throws
IOException {
int sum = 0;
while (values.hasNext()) {
sum += values.next().get();
}
HBaseConfiguration conf = new HBaseConfiguration();
HTable table = new HTable(conf, "phrases");
String row = key.toString();
BatchUpdate update = new BatchUpdate(row);
update.put("counters:value", Bytes.toBytes(sum));
table.commit(update);
}
}
Now as you can see, my reducer has an output collecter of type
<Text,IntWritable> but I don't call the output collector at all. Instead I
load the data to Hbase via table.commit.
I also use NullOutputFormat to avoid getting any empty output files.
This code works and does what I want but I'm not convinced that this is the
right way to do it. I tried to go over the example codes like
BuildTableIndex.java and the others but all of them already had reducers of
the following form:
reduce(ImmutableBytesWritable key, Iterator<RowResult> values,
OutputCollector<ImmutableBytesWritable, LuceneDocumentWrapper> output,
@SuppressWarnings("unused") Reporter reporter)
which is not how I get the intermadiate key,value pairs from the mapper into
the reducer.
Can you give me some advice and a few lines of sample source code if there
is a way to load the data using an output collector? Basically, I'm confused
on how to specify the reducer input parameters and which class to subclass?
I posted this to the mailing list because I couldn't find any more examples
anywhere else.
Thanks in advance,
Jim
Re: Using Hbase as data sink
Posted by stack <st...@duboce.net>.
Jonathan Gray wrote:
> ..
>
> One thing that will really help with performance is to not get a new HTable
> in each map. There was a post in the past day or two regarding this. Pull
> it out into a member of the class, initialize it in the job initialization,
> and just reuse the same one in each reducer task.
>
+1 on above. Can also set up HTable in reducer's configure method.
St.Ack
Re: Using Hbase as data sink
Posted by stack <st...@duboce.net>.
Jim Twensky wrote:
> ...
> I tried to look around to find a method to get the partitioner via JobConf
> but there is no such thing.
Does
http://hadoop.apache.org/core/docs/current/api/org/apache/hadoop/mapred/JobConf.html#setPartitionerClass(java.lang.Class)
not work for you?
St.Ack
Re: Using Hbase as data sink
Posted by Jim Twensky <ji...@gmail.com>.
Hi again,
>>>Pardon me but which 'run' method? Why do you not have access? Its a
public class? (Sorry if I'm missing an obvious -- still on >>>first cup of
coffee).
So here is how my class looks like:
public class PhraseGenerator extends Configured implements Tool {
...
public int run(String[] args) throws Exception {
JobConf conf = new JobConf(getConf(), PhraseGenerator.class);
conf.setJobName("PhraseGenerator");
....
}
}
I tried to look around to find a method to get the partitioner via JobConf
but there is no such thing. At this point, this question is related to
Hadoop rather than Hbase and sorry if I'm asking something too obvious but I
usually check the API documentations and the tutorials before asking
questions and I got stuck.
Thanks,
Jim
On Tue, Dec 23, 2008 at 10:05 AM, stack <st...@duboce.net> wrote:
> Jim Twensky wrote:
>
>> ...
>> Why do we need to set the number of the reduce tasks according to the
>> number
>> of regions? Would it make a performance difference?
>>
>>
>
> Regions are the 'natural' division in hbase. My guess is that the
> partitioner was an attempt at calculating an N for reducers that was other
> than 1 or just some hard-coding.
>
> Other considerations are that at the reduce stage, keys are sorted so
> inserts into hbase will be ordered. In this case, cutting the key space so
> its divided at region boundaries could help distributing the upload and help
> performance. I'd imagine this would work best in a mature table, one that
> is already carrying a load, and where the upload is some smallish percentage
> of the total. Otherwise, regions splitting would throw this partitioner
> calculation out of kilter.
>
> I am asking this because I didn't use it in my implementation. I configure
>> the table name and output formats inside the run method which looks like
>> this:
>>
>> public int run(String[] args) throws Exception {
>> ....
>> conf.setOutputKeyClass(ImmutableBytesWritable.class);
>> conf.setOutputValueClass(BatchUpdate.class);
>> conf.set("output.table.name",args[1]);
>> ...
>>
>> }
>>
>> Notice that I don't have access to the partitioner unlike the
>> initTableReduceJob method. Is there a way to overcome this?
>>
>>
>>
> Pardon me but which 'run' method? Why do you not have access? Its a public
> class? (Sorry if I'm missing an obvious -- still on first cup of coffee).
>
> St.Ack
>
Re: Using Hbase as data sink
Posted by Andrew Purtell <ap...@yahoo.com>.
> From: stack <st...@duboce.net>
> Subject: Re: Using Hbase as data sink
> To: hbase-user@hadoop.apache.org
> Date: Tuesday, December 23, 2008, 8:05 AM
> Jim Twensky wrote:
> > ...
> > Why do we need to set the number of the reduce tasks
> > according to the number of regions? Would it make a
> > performance difference?
> >
>
> Regions are the 'natural' division in hbase. My
> guess is that the partitioner was an attempt at calculating
> an N for reducers that was other than 1 or just some
> hard-coding.
I use the log of the regions as the number of reduce tasks to
run, and the default partitioner which just distributes the
load evenly among the reducers using a hash of the key. More
precisely:
HTable table = new HTable(conf, tableName);
int nrReducers =
(int)Math.ceil(
Math.log1p((double)table.getStartKeys().length));
// ...
job.setNumReduceTasks(nrReducers);
I have no formal reason for this. The method just has a nice
feel to it.
- Andy
Re: Using Hbase as data sink
Posted by stack <st...@duboce.net>.
Jim Twensky wrote:
> ...
> Why do we need to set the number of the reduce tasks according to the number
> of regions? Would it make a performance difference?
>
Regions are the 'natural' division in hbase. My guess is that the
partitioner was an attempt at calculating an N for reducers that was
other than 1 or just some hard-coding.
Other considerations are that at the reduce stage, keys are sorted so
inserts into hbase will be ordered. In this case, cutting the key space
so its divided at region boundaries could help distributing the upload
and help performance. I'd imagine this would work best in a mature
table, one that is already carrying a load, and where the upload is some
smallish percentage of the total. Otherwise, regions splitting would
throw this partitioner calculation out of kilter.
> I am asking this because I didn't use it in my implementation. I configure
> the table name and output formats inside the run method which looks like
> this:
>
> public int run(String[] args) throws Exception {
> ....
>
> conf.setOutputKeyClass(ImmutableBytesWritable.class);
> conf.setOutputValueClass(BatchUpdate.class);
> conf.set("output.table.name",args[1]);
> ...
>
> }
>
> Notice that I don't have access to the partitioner unlike the
> initTableReduceJob method. Is there a way to overcome this?
>
>
Pardon me but which 'run' method? Why do you not have access? Its a
public class? (Sorry if I'm missing an obvious -- still on first cup of
coffee).
St.Ack
Re: Using Hbase as data sink
Posted by Jim Twensky <ji...@gmail.com>.
Thank you, that really helped, I appreciate it. I have a final question
about the following code you posted:
if (partitioner != null) {
job.setPartitionerClass(HRegionPartitioner.class);
HTable outputTable = new HTable(new HBaseConfiguration(job), table);
int regions = outputTable.getRegionsInfo().size();
if (job.getNumReduceTasks() > regions){
job.setNumReduceTasks(outputTable.getRegionsInfo().size())
}
Why do we need to set the number of the reduce tasks according to the number
of regions? Would it make a performance difference?
I am asking this because I didn't use it in my implementation. I configure
the table name and output formats inside the run method which looks like
this:
public int run(String[] args) throws Exception {
....
conf.setOutputKeyClass(ImmutableBytesWritable.class);
conf.setOutputValueClass(BatchUpdate.class);
conf.set("output.table.name",args[1]);
...
}
Notice that I don't have access to the partitioner unlike the
initTableReduceJob method. Is there a way to overcome this?
Thanks
Jim
On Mon, Dec 22, 2008 at 3:43 PM, stack <st...@duboce.net> wrote:
> Jim Twensky wrote:
>
>> Hello Jonathan,
>>
>> Thanks for the fast response. Yes, my question is on other methods to put
>> the same data layout into HBase from my map reduce jobs. I've seen the
>> TableOutputFormat but I couldn't find any example usages of it.
>>
>>
>
> Try src/example/mapred. See SampleUploader. It does like you want with a
> mapper prepping data and then a reducer to insert into hbase table. For how
> the reduce is configured, see the initTableReduceJob in TableMapReduceUtil.
> It looks like this:
>
> job.setOutputFormat(TableOutputFormat.class);
> job.setReducerClass(reducer);
> job.set(TableOutputFormat.OUTPUT_TABLE, table);
> job.setOutputKeyClass(ImmutableBytesWritable.class);
> job.setOutputValueClass(BatchUpdate.class);
> if (partitioner != null) {
> job.setPartitionerClass(HRegionPartitioner.class);
> HTable outputTable = new HTable(new HBaseConfiguration(job), table);
> int regions = outputTable.getRegionsInfo().size();
> if (job.getNumReduceTasks() > regions){
> job.setNumReduceTasks(outputTable.getRegionsInfo().size());
> }
> }
>
> ... sets the output format, reducer, output keys, table name, and tries to
> set reduce count.
>
> If any issue with the above, let us know.
> St.Ack
>
Re: Using Hbase as data sink
Posted by stack <st...@duboce.net>.
Jim Twensky wrote:
> Hello Jonathan,
>
> Thanks for the fast response. Yes, my question is on other methods to put
> the same data layout into HBase from my map reduce jobs. I've seen the
> TableOutputFormat but I couldn't find any example usages of it.
>
Try src/example/mapred. See SampleUploader. It does like you want with
a mapper prepping data and then a reducer to insert into hbase table.
For how the reduce is configured, see the initTableReduceJob in
TableMapReduceUtil. It looks like this:
job.setOutputFormat(TableOutputFormat.class);
job.setReducerClass(reducer);
job.set(TableOutputFormat.OUTPUT_TABLE, table);
job.setOutputKeyClass(ImmutableBytesWritable.class);
job.setOutputValueClass(BatchUpdate.class);
if (partitioner != null) {
job.setPartitionerClass(HRegionPartitioner.class);
HTable outputTable = new HTable(new HBaseConfiguration(job), table);
int regions = outputTable.getRegionsInfo().size();
if (job.getNumReduceTasks() > regions){
job.setNumReduceTasks(outputTable.getRegionsInfo().size());
}
}
... sets the output format, reducer, output keys, table name, and tries
to set reduce count.
If any issue with the above, let us know.
St.Ack
Re: Using Hbase as data sink
Posted by Jim Twensky <ji...@gmail.com>.
Hello Jonathan,
Thanks for the fast response. Yes, my question is on other methods to put
the same data layout into HBase from my map reduce jobs. I've seen the
TableOutputFormat but I couldn't find any example usages of it.
Specifically, when we use the FileOutputFormat, a file called part0000 is
created automatically and the output is written into it. However, when
pulling data into a table, we need to specify the input table name at a
minimum. Also, I think I'll need to use RowResult and ImmutableBytesWritable
objects as input parameters to output.collect but I don't know how to do
that exactly. That's why I was looking over for a few code examples but I
couldn't find much. I'd be glad if you can direct me to some examples or
post here if you have any of your own.
As for the HTable, you are perfectly right. I'll take it outside of the
reducer.
Thanks again,
Jim
On Mon, Dec 22, 2008 at 2:59 PM, Jonathan Gray <jl...@streamy.com> wrote:
> Jim,
>
> This looks like a sane way to do what you want. Is your question strictly
> on other methods to put the same data layout into HBase from the MR job, or
> also about the choice of structure?
>
> As far as how else to use HBase as a data sink, you can make use of
> TableOutputFormat. In my experience, however, it has been faster to
> directly use the API as you are right now. However you can actually batch
> your BatchUpdates automatically using TOF.
>
>
> http://hadoop.apache.org/hbase/docs/current/api/org/apache/hadoop/hbase/mapr
> ed/TableOutputFormat.html<http://hadoop.apache.org/hbase/docs/current/api/org/apache/hadoop/hbase/mapred/TableOutputFormat.html>
>
> One thing that will really help with performance is to not get a new HTable
> in each map. There was a post in the past day or two regarding this. Pull
> it out into a member of the class, initialize it in the job initialization,
> and just reuse the same one in each reducer task.
>
> JG
>
> > -----Original Message-----
> > From: Jim Twensky [mailto:jim.twensky@gmail.com]
> > Sent: Monday, December 22, 2008 12:38 PM
> > To: hbase-user@hadoop.apache.org
> > Subject: Using Hbase as data sink
> >
> > Hello,
> > I have an application which is similar to the word count example given
> > on
> > the Hadoop Map/Reduce tutorial. Instead of counting the words however,
> > I
> > count the phrases. That is, for a sentence like:
> >
> > "How are you"
> >
> > I emit the following phrases inside my mapper:
> >
> > How 1
> > How are 1
> > How are you 1
> > are 1
> > are you 1
> > you 1
> >
> > and then inside the reducer, I aggregate the same keys and send them to
> > an
> > output file.
> >
> > However, I want to load these (phrase,count) pairs to Hbase instead of
> > storing them in a file. I've already written the code and it works but
> > I
> > have some concerns about its performance and I'm not sure if this is
> > the
> > right way to do it. Here is how my reducer looks:
> >
> > public class Reduce extends MapReduceBase implements Reducer<Text,
> > IntWritable, Text, IntWritable> {
> >
> > public void reduce(Text key, Iterator<IntWritable> values,
> > OutputCollector<Text, IntWritable> output, Reporter reporter) throws
> > IOException {
> > int sum = 0;
> > while (values.hasNext()) {
> > sum += values.next().get();
> > }
> >
> > HBaseConfiguration conf = new HBaseConfiguration();
> > HTable table = new HTable(conf, "phrases");
> >
> > String row = key.toString();
> > BatchUpdate update = new BatchUpdate(row);
> > update.put("counters:value", Bytes.toBytes(sum));
> > table.commit(update);
> >
> > }
> > }
> >
> > Now as you can see, my reducer has an output collecter of type
> > <Text,IntWritable> but I don't call the output collector at all.
> > Instead I
> > load the data to Hbase via table.commit.
> > I also use NullOutputFormat to avoid getting any empty output files.
> >
> > This code works and does what I want but I'm not convinced that this is
> > the
> > right way to do it. I tried to go over the example codes like
> > BuildTableIndex.java and the others but all of them already had
> > reducers of
> > the following form:
> >
> > reduce(ImmutableBytesWritable key, Iterator<RowResult> values,
> > OutputCollector<ImmutableBytesWritable, LuceneDocumentWrapper>
> > output,
> > @SuppressWarnings("unused") Reporter reporter)
> >
> > which is not how I get the intermadiate key,value pairs from the mapper
> > into
> > the reducer.
> >
> > Can you give me some advice and a few lines of sample source code if
> > there
> > is a way to load the data using an output collector? Basically, I'm
> > confused
> > on how to specify the reducer input parameters and which class to
> > subclass?
> > I posted this to the mailing list because I couldn't find any more
> > examples
> > anywhere else.
> >
> > Thanks in advance,
> > Jim
>
>
RE: Using Hbase as data sink
Posted by Jonathan Gray <jl...@streamy.com>.
Jim,
This looks like a sane way to do what you want. Is your question strictly
on other methods to put the same data layout into HBase from the MR job, or
also about the choice of structure?
As far as how else to use HBase as a data sink, you can make use of
TableOutputFormat. In my experience, however, it has been faster to
directly use the API as you are right now. However you can actually batch
your BatchUpdates automatically using TOF.
http://hadoop.apache.org/hbase/docs/current/api/org/apache/hadoop/hbase/mapr
ed/TableOutputFormat.html
One thing that will really help with performance is to not get a new HTable
in each map. There was a post in the past day or two regarding this. Pull
it out into a member of the class, initialize it in the job initialization,
and just reuse the same one in each reducer task.
JG
> -----Original Message-----
> From: Jim Twensky [mailto:jim.twensky@gmail.com]
> Sent: Monday, December 22, 2008 12:38 PM
> To: hbase-user@hadoop.apache.org
> Subject: Using Hbase as data sink
>
> Hello,
> I have an application which is similar to the word count example given
> on
> the Hadoop Map/Reduce tutorial. Instead of counting the words however,
> I
> count the phrases. That is, for a sentence like:
>
> "How are you"
>
> I emit the following phrases inside my mapper:
>
> How 1
> How are 1
> How are you 1
> are 1
> are you 1
> you 1
>
> and then inside the reducer, I aggregate the same keys and send them to
> an
> output file.
>
> However, I want to load these (phrase,count) pairs to Hbase instead of
> storing them in a file. I've already written the code and it works but
> I
> have some concerns about its performance and I'm not sure if this is
> the
> right way to do it. Here is how my reducer looks:
>
> public class Reduce extends MapReduceBase implements Reducer<Text,
> IntWritable, Text, IntWritable> {
>
> public void reduce(Text key, Iterator<IntWritable> values,
> OutputCollector<Text, IntWritable> output, Reporter reporter) throws
> IOException {
> int sum = 0;
> while (values.hasNext()) {
> sum += values.next().get();
> }
>
> HBaseConfiguration conf = new HBaseConfiguration();
> HTable table = new HTable(conf, "phrases");
>
> String row = key.toString();
> BatchUpdate update = new BatchUpdate(row);
> update.put("counters:value", Bytes.toBytes(sum));
> table.commit(update);
>
> }
> }
>
> Now as you can see, my reducer has an output collecter of type
> <Text,IntWritable> but I don't call the output collector at all.
> Instead I
> load the data to Hbase via table.commit.
> I also use NullOutputFormat to avoid getting any empty output files.
>
> This code works and does what I want but I'm not convinced that this is
> the
> right way to do it. I tried to go over the example codes like
> BuildTableIndex.java and the others but all of them already had
> reducers of
> the following form:
>
> reduce(ImmutableBytesWritable key, Iterator<RowResult> values,
> OutputCollector<ImmutableBytesWritable, LuceneDocumentWrapper>
> output,
> @SuppressWarnings("unused") Reporter reporter)
>
> which is not how I get the intermadiate key,value pairs from the mapper
> into
> the reducer.
>
> Can you give me some advice and a few lines of sample source code if
> there
> is a way to load the data using an output collector? Basically, I'm
> confused
> on how to specify the reducer input parameters and which class to
> subclass?
> I posted this to the mailing list because I couldn't find any more
> examples
> anywhere else.
>
> Thanks in advance,
> Jim