Coprocessors and batch processing

[lars hofhansl <lh...@yahoo.com>]

Here's another coprocessor question...

>From the client we batch operations in order to reduce the number of round trips.
Currently there is no way (that I can find) to make use of those batches in coprocessors.

This is an issue when, for example, sets of puts and gets are (partially) forwarded to another table by the coprocessor.
Right now this would need to use many single puts/deletes/gets from the various {pre|post}{put|delete|get} hooks.

There is no useful demarcation; other than maybe waiting a few miliseconds, which is awkward.


Of course this forwarding could be done directly from the client, put then what's the point of coprocessors?

I guess there could either be a {pre|post}Multi on RegionObserver (although HRegionServer.multi does a lot of munging).
Or maybe a general {pre|post}Request with no arguments - in which case it would be at least possible to write code in the coprocessor
to collect the puts/deletes/etc through the normal single prePut/preDelete/etc hooks and then batch-process them in postRequest().

-- Lars

Re: Coprocessors and batch processing

[lars hofhansl <lh...@yahoo.com>]

Thanks Gary.

Created https://issues.apache.org/jira/browse/HBASE-4210 for discussion.



________________________________
From: Gary Helmling <gh...@gmail.com>
To: user@hbase.apache.org; lars hofhansl <lh...@yahoo.com>
Sent: Monday, August 15, 2011 11:35 PM
Subject: Re: Coprocessors and batch processing

Hi Lars,

Should then all RPC triggered by a coprocessor be avoided (and hence the use
> the env-provided HTableInterface be generally discouraged)?
>
>
I would generally avoid making synchronous RPC calls within a direct
coprocessor call path (blocking a handler thread waiting on the response).
Making the same calls asynchronously (say queuing puts for a secondary index
to be handled by a background thread) is generally better because you're not
tying up a constrained resource that other clients will be contending on.

The cp environment provided HTableInterface is there to be used, but it's
still up to you to use it wisely.  But providing that resource as part of
the coprocessor framework (instead of using a standard client HTable) will
potentially allow us to do other optimizations like using a separate
priority for coprocessor originating RPCs (handled by a different thread
pool than client RPCs), or short circuiting the RPC stack for calls to
regions residing on the same region server (and avoiding tying up another
handler thread).

Those are just a couple examples, but I think ultimately we'll want a bit
more constraint over what coprocessor code is allowed to do, for the sake of
better guaranteeing cluster stability.  Currently it's a void-your-warranty
type scenario. :)

I still think a RegionServer or RpcServer level "preRequest" and
> "postRequest" (or whatever) hooks would be useful for a variety of
> scenarios.
>
>
I could see that, but it's definitely not part of the RegionObserver
contract.  I do also worry that a profusion of too many different
coprocessor interfaces will lead to confusion about how to actually go about
implementing a given application's needs.  But we're still pretty early on
in the development of coprocessors and I don't pretend we have everything
covered.

Do you have some specific scenarios where you think a preRequest/postRequest
interface would be better suited?  Feel free to open up a JIRA where we can
walk through them!  We could try to model out a RPC listener or RPC filter
interface.  I think that interacting at the RPC layer (in front of all of
the core HBase code) will be a bit limited in what you have access to.  Many
of the current coprocessor hooks are situated in keys points after a lot of
setup or initialization has gone on to provide the call context.  But for a
given set of needs that may not be an issue (or may even be an advantage).


Gary

Re: Coprocessors and batch processing

[Gary Helmling <gh...@gmail.com>]

Hi Lars,

Should then all RPC triggered by a coprocessor be avoided (and hence the use
> the env-provided HTableInterface be generally discouraged)?
>
>
I would generally avoid making synchronous RPC calls within a direct
coprocessor call path (blocking a handler thread waiting on the response).
Making the same calls asynchronously (say queuing puts for a secondary index
to be handled by a background thread) is generally better because you're not
tying up a constrained resource that other clients will be contending on.

The cp environment provided HTableInterface is there to be used, but it's
still up to you to use it wisely.  But providing that resource as part of
the coprocessor framework (instead of using a standard client HTable) will
potentially allow us to do other optimizations like using a separate
priority for coprocessor originating RPCs (handled by a different thread
pool than client RPCs), or short circuiting the RPC stack for calls to
regions residing on the same region server (and avoiding tying up another
handler thread).

Those are just a couple examples, but I think ultimately we'll want a bit
more constraint over what coprocessor code is allowed to do, for the sake of
better guaranteeing cluster stability.  Currently it's a void-your-warranty
type scenario. :)

I still think a RegionServer or RpcServer level "preRequest" and
> "postRequest" (or whatever) hooks would be useful for a variety of
> scenarios.
>
>
I could see that, but it's definitely not part of the RegionObserver
contract.  I do also worry that a profusion of too many different
coprocessor interfaces will lead to confusion about how to actually go about
implementing a given application's needs.  But we're still pretty early on
in the development of coprocessors and I don't pretend we have everything
covered.

Do you have some specific scenarios where you think a preRequest/postRequest
interface would be better suited?  Feel free to open up a JIRA where we can
walk through them!  We could try to model out a RPC listener or RPC filter
interface.  I think that interacting at the RPC layer (in front of all of
the core HBase code) will be a bit limited in what you have access to.  Many
of the current coprocessor hooks are situated in keys points after a lot of
setup or initialization has gone on to provide the call context.  But for a
given set of needs that may not be an issue (or may even be an advantage).


Gary

Re: Coprocessors and batch processing

[lars hofhansl <lh...@yahoo.com>]

Thanks Gary (and Himanshu),

correct, using batch operations is just an optimization and not a semantic difference to single row ops.


Should then all RPC triggered by a coprocessor be avoided (and hence the use the env-provided HTableInterface be generally discouraged)?

The 2ndary index scenario (with separate tables) was just a "bad" example, which is better handled at the client anyway.


I still think a RegionServer or RpcServer level "preRequest" and "postRequest" (or whatever) hooks would be useful for a variety of scenarios.


Thanks.

-- Lars



________________________________
From: Gary Helmling <gh...@gmail.com>
To: user@hbase.apache.org
Sent: Thursday, August 11, 2011 11:24 AM
Subject: Re: Coprocessors and batch processing

On Wed, Aug 10, 2011 at 10:46 PM, lars hofhansl <lh...@yahoo.com> wrote:

>
> I guess there could either be a {pre|post}Multi on RegionObserver (although
> HRegionServer.multi does a lot of munging).
> Or maybe a general {pre|post}Request with no arguments - in which case it
> would be at least possible to write code in the coprocessor
> to collect the puts/deletes/etc through the normal single
> prePut/preDelete/etc hooks and then batch-process them in postRequest().
>
>
This same question came up recently in an internal discussion as well.

Since multi() only exists at the HRegionServer level (there is no
HRegion.multi()), I don't think that this belongs in RegionObserver at all.
So this would be the main reason why there currently are no
preMulti()/postMulti() hooks.  There is something of a gap here between the
way the client sees things and the way the coprocessor sees things.  But I
really see this as more of an RPC-layer listener or filter.  After all,
"multi" is really just an RPC operation for efficiency.  It's not really a
core HBase operation.

As I see it there are a couple motivations behind the current limitation:

1) we generally only want a single set of coprocessor hooks to be involved
in a given operation
2) the coprocessor API should be as conceptually simple as possible, while
still reflecting what is happening in HBase


I think adding in some multi representation in the coprocessor API poses
challenges on each of these fronts.  Not necessarily unresolvable
challenges, but there are trade-offs involved.

Re (1): representing multi in cp hooks means that you now have layering in
the cp hooks handling a given operation.  Say all the actions in the multi
request are Puts, you go from having:

[prePut() ... postPut()] x N

to

preMulti() [prePut() ... postPut()] x N  postMulti()

For me, this implies some confusion about where I should handle the actions
in my coprocessor.  Should I put all handling in pre/postMulti() or do I
also need to implement pre/postPut()?

This gets at (2), for me it's easier to think of "multi" as just an
aggregation of operations, not as an operation in itself.  The actual
operation in the above example is a Put.  It's just that there are a lot of
them.


Back to the original situation you raise, I think it's a really bad idea to
immediately trigger RPC operations within a coprocessor, _especially_ in the
case of multi.  Say you are doing a secondary indexing transformation on the
Puts you receive.  You get a multi batch of 1000 puts.  You transform that
into a batch of 1000 secondary index puts, potentially going to every region
server in your cluster holding a region in the secondary indexing table.
You've just multiplied the RPC operations triggered by a single request and
exposed yourself to triggering a distributed deadlock, where the RPC handler
thread running in one RS is waiting for an RPC handler to become available
in another RS, which in turn has all handlers occupied waiting on other
servers.

I think the better approach to doing these kind of updates would be to have
the RegionObserver.pre/postPut() implementation queue them up and have them
batched and processed by a separate background thread so that you're not
tying up resources directly in the RPC handling path (and also making
clients wait on a synchronous response).

It may be that a higher level (RegionServer or RpcServer level)
observer-type interface would be useful.  But I think that adds some
complexity to understanding what coprocessors are and how they interact.

--gh

Re: Coprocessors and batch processing

[Gary Helmling <gh...@gmail.com>]

On Wed, Aug 10, 2011 at 10:46 PM, lars hofhansl <lh...@yahoo.com> wrote:

>
> I guess there could either be a {pre|post}Multi on RegionObserver (although
> HRegionServer.multi does a lot of munging).
> Or maybe a general {pre|post}Request with no arguments - in which case it
> would be at least possible to write code in the coprocessor
> to collect the puts/deletes/etc through the normal single
> prePut/preDelete/etc hooks and then batch-process them in postRequest().
>
>
This same question came up recently in an internal discussion as well.

Since multi() only exists at the HRegionServer level (there is no
HRegion.multi()), I don't think that this belongs in RegionObserver at all.
So this would be the main reason why there currently are no
preMulti()/postMulti() hooks.  There is something of a gap here between the
way the client sees things and the way the coprocessor sees things.  But I
really see this as more of an RPC-layer listener or filter.  After all,
"multi" is really just an RPC operation for efficiency.  It's not really a
core HBase operation.

As I see it there are a couple motivations behind the current limitation:

1) we generally only want a single set of coprocessor hooks to be involved
in a given operation
2) the coprocessor API should be as conceptually simple as possible, while
still reflecting what is happening in HBase


I think adding in some multi representation in the coprocessor API poses
challenges on each of these fronts.  Not necessarily unresolvable
challenges, but there are trade-offs involved.

Re (1): representing multi in cp hooks means that you now have layering in
the cp hooks handling a given operation.  Say all the actions in the multi
request are Puts, you go from having:

[prePut() ... postPut()] x N

to

preMulti() [prePut() ... postPut()] x N  postMulti()

For me, this implies some confusion about where I should handle the actions
in my coprocessor.  Should I put all handling in pre/postMulti() or do I
also need to implement pre/postPut()?

This gets at (2), for me it's easier to think of "multi" as just an
aggregation of operations, not as an operation in itself.  The actual
operation in the above example is a Put.  It's just that there are a lot of
them.


Back to the original situation you raise, I think it's a really bad idea to
immediately trigger RPC operations within a coprocessor, _especially_ in the
case of multi.  Say you are doing a secondary indexing transformation on the
Puts you receive.  You get a multi batch of 1000 puts.  You transform that
into a batch of 1000 secondary index puts, potentially going to every region
server in your cluster holding a region in the secondary indexing table.
You've just multiplied the RPC operations triggered by a single request and
exposed yourself to triggering a distributed deadlock, where the RPC handler
thread running in one RS is waiting for an RPC handler to become available
in another RS, which in turn has all handlers occupied waiting on other
servers.

I think the better approach to doing these kind of updates would be to have
the RegionObserver.pre/postPut() implementation queue them up and have them
batched and processed by a separate background thread so that you're not
tying up resources directly in the RPC handling path (and also making
clients wait on a synchronous response).

It may be that a higher level (RegionServer or RpcServer level)
observer-type interface would be useful.  But I think that adds some
complexity to understanding what coprocessors are and how they interact.

--gh

Re: Coprocessors and batch processing

[Himanshu Vashishtha <hv...@cs.ualberta.ca>]

Hey Lars,
Sorry if I have mislead you.

The current Coprocessor infrastructure is at _Region_ level, not at
_RegionServer_ level.
All these batch operations are ultimately ends up at some rows in some
Regions, where you have hooked your CPs.

I am not able to follow your example. If you end up with a 200 Puts batch at
a RS, then what? You need to execute this at Region level now, right? A RS
just host Regions, and these Regions are mobile too. I think that is the
reason one should nail down to a specific Region while doing any crud
operations here; and also a reason why current CPs are at Region level.

That's my pov though! :)

Himanshu




On Thu, Aug 11, 2011 at 11:56 AM, lars hofhansl <lh...@yahoo.com> wrote:

> Thanks Himanshu,
>
> but that is not quite what I meant.
>
>
> Yes, a batch operation is broken up in "chunks" per regionserver and then
> the chunks are shipped to the individual regionservers.
> But then there is no way to interact with those chunks at the regionserver
> through coprocessors(as a whole).
>
>
> What I want to do is to look at the entire chunk at each regionserver and
> then do some other bulk operation based on that.
> Currently I only get pre/post hooks for single rows, and no way to group
> these together later (other than just waiting for a little bit and let
> work accumulate).
>
> Say I have a client request with (say) 1000 puts, and let's also say that
> there are 5 region server, and each happens to host exactly 1/5th of the
> rowkeys, so each region server gets a chunk of 200 puts.
> Now a coprocessor might have logic that affect another table (for example
> for naive 2ndary indexing). At the coprocessor level I can get an
>
> HTableInterface from the environment and now I want to do a batch put of
> 200 rows (of course those will be broken up per region server again, etc).
> Currently I can't do that, because there are only single "row" pre/post
> hooks, and no way to determine when all operations of a request are done.
> The end result is that I have to do 200 single row puts, one in each call to
> pre or post hooks.
>
>
> Does that make sense?
>
> -- Lars
>
>
>
> ________________________________
> From: Himanshu Vashishtha <hv...@cs.ualberta.ca>
> To: user@hbase.apache.org; lars hofhansl <lh...@yahoo.com>
> Sent: Wednesday, August 10, 2011 11:21 PM
> Subject: Re: Coprocessors and batch processing
>
> Client side batch processing is done at RegionServer level, i.e., all
> Action
> objects are grouped together per RS basis and send in one RPC. Once the
> batch arrives at a RS, it gets distributed across corresponding Regions,
> and
> these Action objects are processed, one by one. This include Coprocessor's
> Exec objects too.
> So, a coprocessor is working at a "Region" level granularity.
>
> If you want to take some action (process bunch of rows of another table
> from
> a CP), one can get a HTable instance from Environment instance of a
> Coprocessor, and use the same mechanism as used by the client side.
> Will that help in your use-case?
>
> Thanks,
> Himanshu
>
>
> On Wed, Aug 10, 2011 at 11:46 PM, lars hofhansl <lh...@yahoo.com>
> wrote:
>
> > Here's another coprocessor question...
> >
> > From the client we batch operations in order to reduce the number of
> round
> > trips.
> > Currently there is no way (that I can find) to make use of those batches
> in
> > coprocessors.
> >
> > This is an issue when, for example, sets of puts and gets are (partially)
> > forwarded to another table by the coprocessor.
> > Right now this would need to use many single puts/deletes/gets from the
> > various {pre|post}{put|delete|get} hooks.
> >
> > There is no useful demarcation; other than maybe waiting a few
> miliseconds,
> > which is awkward.
> >
> >
> > Of course this forwarding could be done directly from the client, put
> then
> > what's the point of coprocessors?
> >
> > I guess there could either be a {pre|post}Multi on RegionObserver
> (although
> > HRegionServer.multi does a lot of munging).
> > Or maybe a general {pre|post}Request with no arguments - in which case it
> > would be at least possible to write code in the coprocessor
> > to collect the puts/deletes/etc through the normal single
> > prePut/preDelete/etc hooks and then batch-process them in postRequest().
> >
> > -- Lars
> >
>

Re: Coprocessors and batch processing

[lars hofhansl <lh...@yahoo.com>]

Thanks Himanshu,

but that is not quite what I meant.


Yes, a batch operation is broken up in "chunks" per regionserver and then the chunks are shipped to the individual regionservers.
But then there is no way to interact with those chunks at the regionserver through coprocessors(as a whole).


What I want to do is to look at the entire chunk at each regionserver and then do some other bulk operation based on that.
Currently I only get pre/post hooks for single rows, and no way to group these together later (other than just waiting for a little bit and let
work accumulate).

Say I have a client request with (say) 1000 puts, and let's also say that there are 5 region server, and each happens to host exactly 1/5th of the rowkeys, so each region server gets a chunk of 200 puts.
Now a coprocessor might have logic that affect another table (for example for naive 2ndary indexing). At the coprocessor level I can get an 

HTableInterface from the environment and now I want to do a batch put of 200 rows (of course those will be broken up per region server again, etc).
Currently I can't do that, because there are only single "row" pre/post hooks, and no way to determine when all operations of a request are done. The end result is that I have to do 200 single row puts, one in each call to pre or post hooks.


Does that make sense?

-- Lars



________________________________
From: Himanshu Vashishtha <hv...@cs.ualberta.ca>
To: user@hbase.apache.org; lars hofhansl <lh...@yahoo.com>
Sent: Wednesday, August 10, 2011 11:21 PM
Subject: Re: Coprocessors and batch processing

Client side batch processing is done at RegionServer level, i.e., all Action
objects are grouped together per RS basis and send in one RPC. Once the
batch arrives at a RS, it gets distributed across corresponding Regions, and
these Action objects are processed, one by one. This include Coprocessor's
Exec objects too.
So, a coprocessor is working at a "Region" level granularity.

If you want to take some action (process bunch of rows of another table from
a CP), one can get a HTable instance from Environment instance of a
Coprocessor, and use the same mechanism as used by the client side.
Will that help in your use-case?

Thanks,
Himanshu


On Wed, Aug 10, 2011 at 11:46 PM, lars hofhansl <lh...@yahoo.com> wrote:

> Here's another coprocessor question...
>
> From the client we batch operations in order to reduce the number of round
> trips.
> Currently there is no way (that I can find) to make use of those batches in
> coprocessors.
>
> This is an issue when, for example, sets of puts and gets are (partially)
> forwarded to another table by the coprocessor.
> Right now this would need to use many single puts/deletes/gets from the
> various {pre|post}{put|delete|get} hooks.
>
> There is no useful demarcation; other than maybe waiting a few miliseconds,
> which is awkward.
>
>
> Of course this forwarding could be done directly from the client, put then
> what's the point of coprocessors?
>
> I guess there could either be a {pre|post}Multi on RegionObserver (although
> HRegionServer.multi does a lot of munging).
> Or maybe a general {pre|post}Request with no arguments - in which case it
> would be at least possible to write code in the coprocessor
> to collect the puts/deletes/etc through the normal single
> prePut/preDelete/etc hooks and then batch-process them in postRequest().
>
> -- Lars
>

Re: Coprocessors and batch processing

[Xian Woo <in...@gmail.com>]

Excuse me, I‘ve got a question here. Can we use coprocessor in hbase-0.90-*?
I've heard of  a saying that we can't use coprocessor until
the hbase-0.91-*. And is there any possible way to get  the hbase-0.91-*
release or even after that?
Thanks.
With regards.

Woo

2011/8/11 Himanshu Vashishtha <hv...@cs.ualberta.ca>

> Client side batch processing is done at RegionServer level, i.e., all
> Action
> objects are grouped together per RS basis and send in one RPC. Once the
> batch arrives at a RS, it gets distributed across corresponding Regions,
> and
> these Action objects are processed, one by one. This include Coprocessor's
> Exec objects too.
> So, a coprocessor is working at a "Region" level granularity.
>
> If you want to take some action (process bunch of rows of another table
> from
> a CP), one can get a HTable instance from Environment instance of a
> Coprocessor, and use the same mechanism as used by the client side.
> Will that help in your use-case?
>
> Thanks,
> Himanshu
>
>
> On Wed, Aug 10, 2011 at 11:46 PM, lars hofhansl <lh...@yahoo.com>
> wrote:
>
> > Here's another coprocessor question...
> >
> > From the client we batch operations in order to reduce the number of
> round
> > trips.
> > Currently there is no way (that I can find) to make use of those batches
> in
> > coprocessors.
> >
> > This is an issue when, for example, sets of puts and gets are (partially)
> > forwarded to another table by the coprocessor.
> > Right now this would need to use many single puts/deletes/gets from the
> > various {pre|post}{put|delete|get} hooks.
> >
> > There is no useful demarcation; other than maybe waiting a few
> miliseconds,
> > which is awkward.
> >
> >
> > Of course this forwarding could be done directly from the client, put
> then
> > what's the point of coprocessors?
> >
> > I guess there could either be a {pre|post}Multi on RegionObserver
> (although
> > HRegionServer.multi does a lot of munging).
> > Or maybe a general {pre|post}Request with no arguments - in which case it
> > would be at least possible to write code in the coprocessor
> > to collect the puts/deletes/etc through the normal single
> > prePut/preDelete/etc hooks and then batch-process them in postRequest().
> >
> > -- Lars
> >
>

Re: Coprocessors and batch processing

[Himanshu Vashishtha <hv...@cs.ualberta.ca>]

Client side batch processing is done at RegionServer level, i.e., all Action
objects are grouped together per RS basis and send in one RPC. Once the
batch arrives at a RS, it gets distributed across corresponding Regions, and
these Action objects are processed, one by one. This include Coprocessor's
Exec objects too.
So, a coprocessor is working at a "Region" level granularity.

If you want to take some action (process bunch of rows of another table from
a CP), one can get a HTable instance from Environment instance of a
Coprocessor, and use the same mechanism as used by the client side.
Will that help in your use-case?

Thanks,
Himanshu


On Wed, Aug 10, 2011 at 11:46 PM, lars hofhansl <lh...@yahoo.com> wrote:

> Here's another coprocessor question...
>
> From the client we batch operations in order to reduce the number of round
> trips.
> Currently there is no way (that I can find) to make use of those batches in
> coprocessors.
>
> This is an issue when, for example, sets of puts and gets are (partially)
> forwarded to another table by the coprocessor.
> Right now this would need to use many single puts/deletes/gets from the
> various {pre|post}{put|delete|get} hooks.
>
> There is no useful demarcation; other than maybe waiting a few miliseconds,
> which is awkward.
>
>
> Of course this forwarding could be done directly from the client, put then
> what's the point of coprocessors?
>
> I guess there could either be a {pre|post}Multi on RegionObserver (although
> HRegionServer.multi does a lot of munging).
> Or maybe a general {pre|post}Request with no arguments - in which case it
> would be at least possible to write code in the coprocessor
> to collect the puts/deletes/etc through the normal single
> prePut/preDelete/etc hooks and then batch-process them in postRequest().
>
> -- Lars
>