We are using a 6-node HBase cluster with a Thrift Server on each of the RegionServer nodes, and trying to evaluate maximum write throughput for our use case (which involves many processes sending mutateRowsTs commands). Somewhere between about 30 and 40 processes writing into the system we cross the threshold where adding additional writers yields only very limited returns to throughput, and I'm not sure why. We see that the CPU and Disk on the DataNode/RegionServer/ThriftServer machines are not saturated, nor is the NIC in those machines. I'm a little unsure where to look next.

A little more detail about our deployment:

* CDH 4.1.2
* DataNode/RegionServer/ThriftServer class: EC2 m1.xlarge
** RegionServer: 8GB heap
** ThriftServer: 1GB heap
** DataNode: 4GB heap
** EC2 ephemeral (i.e. local, not EBS) volumes used for HDFS

If there's any other information that I can provide, or any other configuration or system settings I should look at, I'd appreciate the pointers.

Thanks,
 - Dan

Maybe you've the limit of the number of RPC Threads handling your write
requests per Region Server, thus your requests are queued?
On Fri, Mar 1, 2013 at 2:17 PM, Dan Crosta <[EMAIL PROTECTED]> wrote:

> We are using a 6-node HBase cluster with a Thrift Server on each of the
> RegionServer nodes, and trying to evaluate maximum write throughput for our
> use case (which involves many processes sending mutateRowsTs commands).
> Somewhere between about 30 and 40 processes writing into the system we
> cross the threshold where adding additional writers yields only very
> limited returns to throughput, and I'm not sure why. We see that the CPU
> and Disk on the DataNode/RegionServer/ThriftServer machines are not
> saturated, nor is the NIC in those machines. I'm a little unsure where to
> look next.
>
> A little more detail about our deployment:
>
> * CDH 4.1.2
> * DataNode/RegionServer/ThriftServer class: EC2 m1.xlarge
> ** RegionServer: 8GB heap
> ** ThriftServer: 1GB heap
> ** DataNode: 4GB heap
> ** EC2 ephemeral (i.e. local, not EBS) volumes used for HDFS
>
> If there's any other information that I can provide, or any other
> configuration or system settings I should look at, I'd appreciate the
> pointers.
>
> Thanks,
>  - Dan

On Mar 1, 2013, at 9:13 AM, Asaf Mesika wrote:
> Maybe you've the limit of the number of RPC Threads handling your write
> requests per Region Server, thus your requests are queued?

Thanks -- I actually just stumbled across that setting in CDH manager, and am experimenting with raising it.

The HBase book warns "The rule of thumb is to keep this number low when the payload per request approaches the MB (big puts, scans using a large cache) and high when the payload is small (gets, small puts, ICVs, deletes)" -- we do have a pattern of lots of small writes, and infrequent and relatively few large reads. Should I be worried about this?

Thanks,
- Dan

The primary unit of load distribution in HBase is the region, make
sure you have more than one. This is well documented in the manual
http://hbase.apache.org/book/perf.writing.html

J-D

On Fri, Mar 1, 2013 at 4:17 AM, Dan Crosta <[EMAIL PROTECTED]> wrote:
> We are using a 6-node HBase cluster with a Thrift Server on each of the RegionServer nodes, and trying to evaluate maximum write throughput for our use case (which involves many processes sending mutateRowsTs commands). Somewhere between about 30 and 40 processes writing into the system we cross the threshold where adding additional writers yields only very limited returns to throughput, and I'm not sure why. We see that the CPU and Disk on the DataNode/RegionServer/ThriftServer machines are not saturated, nor is the NIC in those machines. I'm a little unsure where to look next.
>
> A little more detail about our deployment:
>
> * CDH 4.1.2
> * DataNode/RegionServer/ThriftServer class: EC2 m1.xlarge
> ** RegionServer: 8GB heap
> ** ThriftServer: 1GB heap
> ** DataNode: 4GB heap
> ** EC2 ephemeral (i.e. local, not EBS) volumes used for HDFS
>
> If there's any other information that I can provide, or any other configuration or system settings I should look at, I'd appreciate the pointers.
>
> Thanks,
>  - Dan

Hi,

I don't know how many worker threads you have at the thrift servers. Each
thread gets dedicated to a single connection and only serves that
connection. New connections get queued. Also, are you sure that you are not
saturating the client side making the calls ?

Varun

On Fri, Mar 1, 2013 at 9:33 AM, Jean-Daniel Cryans <[EMAIL PROTECTED]>wrote:

> The primary unit of load distribution in HBase is the region, make
> sure you have more than one. This is well documented in the manual
> http://hbase.apache.org/book/perf.writing.html
>
> J-D
>
> On Fri, Mar 1, 2013 at 4:17 AM, Dan Crosta <[EMAIL PROTECTED]> wrote:
> > We are using a 6-node HBase cluster with a Thrift Server on each of the
> RegionServer nodes, and trying to evaluate maximum write throughput for our
> use case (which involves many processes sending mutateRowsTs commands).
> Somewhere between about 30 and 40 processes writing into the system we
> cross the threshold where adding additional writers yields only very
> limited returns to throughput, and I'm not sure why. We see that the CPU
> and Disk on the DataNode/RegionServer/ThriftServer machines are not
> saturated, nor is the NIC in those machines. I'm a little unsure where to
> look next.
> >
> > A little more detail about our deployment:
> >
> > * CDH 4.1.2
> > * DataNode/RegionServer/ThriftServer class: EC2 m1.xlarge
> > ** RegionServer: 8GB heap
> > ** ThriftServer: 1GB heap
> > ** DataNode: 4GB heap
> > ** EC2 ephemeral (i.e. local, not EBS) volumes used for HDFS
> >
> > If there's any other information that I can provide, or any other
> configuration or system settings I should look at, I'd appreciate the
> pointers.
> >
> > Thanks,
> >  - Dan
>

Did you try running 30-40 proc(s) on one machine and another 30-40 proc(s)
on another machine to see if that doubles the throughput ?

On Fri, Mar 1, 2013 at 10:46 AM, Varun Sharma <[EMAIL PROTECTED]> wrote:

> Hi,
>
> I don't know how many worker threads you have at the thrift servers. Each
> thread gets dedicated to a single connection and only serves that
> connection. New connections get queued. Also, are you sure that you are not
> saturating the client side making the calls ?
>
> Varun
>
>
> On Fri, Mar 1, 2013 at 9:33 AM, Jean-Daniel Cryans <[EMAIL PROTECTED]>wrote:
>
>> The primary unit of load distribution in HBase is the region, make
>> sure you have more than one. This is well documented in the manual
>> http://hbase.apache.org/book/perf.writing.html
>>
>> J-D
>>
>> On Fri, Mar 1, 2013 at 4:17 AM, Dan Crosta <[EMAIL PROTECTED]> wrote:
>> > We are using a 6-node HBase cluster with a Thrift Server on each of the
>> RegionServer nodes, and trying to evaluate maximum write throughput for our
>> use case (which involves many processes sending mutateRowsTs commands).
>> Somewhere between about 30 and 40 processes writing into the system we
>> cross the threshold where adding additional writers yields only very
>> limited returns to throughput, and I'm not sure why. We see that the CPU
>> and Disk on the DataNode/RegionServer/ThriftServer machines are not
>> saturated, nor is the NIC in those machines. I'm a little unsure where to
>> look next.
>> >
>> > A little more detail about our deployment:
>> >
>> > * CDH 4.1.2
>> > * DataNode/RegionServer/ThriftServer class: EC2 m1.xlarge
>> > ** RegionServer: 8GB heap
>> > ** ThriftServer: 1GB heap
>> > ** DataNode: 4GB heap
>> > ** EC2 ephemeral (i.e. local, not EBS) volumes used for HDFS
>> >
>> > If there's any other information that I can provide, or any other
>> configuration or system settings I should look at, I'd appreciate the
>> pointers.
>> >
>> > Thanks,
>> >  - Dan
>>
>
>

We are generating the load from multiple machines, yes.

Do you happen to know what the name of the setting for the number of ThriftServer threads is called? I can't find anything that is obviously about that in the CDH manager.

- Dan
On Mar 1, 2013, at 1:46 PM, Varun Sharma wrote:

> Did you try running 30-40 proc(s) on one machine and another 30-40 proc(s)
> on another machine to see if that doubles the throughput ?
>
> On Fri, Mar 1, 2013 at 10:46 AM, Varun Sharma <[EMAIL PROTECTED]> wrote:
>
>> Hi,
>>
>> I don't know how many worker threads you have at the thrift servers. Each
>> thread gets dedicated to a single connection and only serves that
>> connection. New connections get queued. Also, are you sure that you are not
>> saturating the client side making the calls ?
>>
>> Varun
>>
>>
>> On Fri, Mar 1, 2013 at 9:33 AM, Jean-Daniel Cryans <[EMAIL PROTECTED]>wrote:
>>
>>> The primary unit of load distribution in HBase is the region, make
>>> sure you have more than one. This is well documented in the manual
>>> http://hbase.apache.org/book/perf.writing.html
>>>
>>> J-D
>>>
>>> On Fri, Mar 1, 2013 at 4:17 AM, Dan Crosta <[EMAIL PROTECTED]> wrote:
>>>> We are using a 6-node HBase cluster with a Thrift Server on each of the
>>> RegionServer nodes, and trying to evaluate maximum write throughput for our
>>> use case (which involves many processes sending mutateRowsTs commands).
>>> Somewhere between about 30 and 40 processes writing into the system we
>>> cross the threshold where adding additional writers yields only very
>>> limited returns to throughput, and I'm not sure why. We see that the CPU
>>> and Disk on the DataNode/RegionServer/ThriftServer machines are not
>>> saturated, nor is the NIC in those machines. I'm a little unsure where to
>>> look next.
>>>>
>>>> A little more detail about our deployment:
>>>>
>>>> * CDH 4.1.2
>>>> * DataNode/RegionServer/ThriftServer class: EC2 m1.xlarge
>>>> ** RegionServer: 8GB heap
>>>> ** ThriftServer: 1GB heap
>>>> ** DataNode: 4GB heap
>>>> ** EC2 ephemeral (i.e. local, not EBS) volumes used for HDFS
>>>>
>>>> If there's any other information that I can provide, or any other
>>> configuration or system settings I should look at, I'd appreciate the
>>> pointers.
>>>>
>>>> Thanks,
>>>> - Dan
>>>
>>
>>

Here're the parameters you should look at:

hbase-server/src/main/java/org/apache/hadoop/hbase/thrift/HThreadedSelectorServerArgs.java:
     "hbase.thrift.selector.threads";
hbase-server/src/main/java/org/apache/hadoop/hbase/thrift/HThreadedSelectorServerArgs.java:
     "hbase.thrift.worker.threads";
hbase-server/src/main/java/org/apache/hadoop/hbase/thrift/TBoundedThreadPoolServer.java:
     "hbase.thrift.threadKeepAliveTimeSec";

Often times, source code is the best help :-)

On Fri, Mar 1, 2013 at 10:49 AM, Dan Crosta <[EMAIL PROTECTED]> wrote:

> We are generating the load from multiple machines, yes.
>
> Do you happen to know what the name of the setting for the number of
> ThriftServer threads is called? I can't find anything that is obviously
> about that in the CDH manager.
>
> - Dan
>
>
> On Mar 1, 2013, at 1:46 PM, Varun Sharma wrote:
>
> > Did you try running 30-40 proc(s) on one machine and another 30-40
> proc(s)
> > on another machine to see if that doubles the throughput ?
> >
> > On Fri, Mar 1, 2013 at 10:46 AM, Varun Sharma <[EMAIL PROTECTED]>
> wrote:
> >
> >> Hi,
> >>
> >> I don't know how many worker threads you have at the thrift servers.
> Each
> >> thread gets dedicated to a single connection and only serves that
> >> connection. New connections get queued. Also, are you sure that you are
> not
> >> saturating the client side making the calls ?
> >>
> >> Varun
> >>
> >>
> >> On Fri, Mar 1, 2013 at 9:33 AM, Jean-Daniel Cryans <[EMAIL PROTECTED]
> >wrote:
> >>
> >>> The primary unit of load distribution in HBase is the region, make
> >>> sure you have more than one. This is well documented in the manual
> >>> http://hbase.apache.org/book/perf.writing.html
> >>>
> >>> J-D
> >>>
> >>> On Fri, Mar 1, 2013 at 4:17 AM, Dan Crosta <[EMAIL PROTECTED]> wrote:
> >>>> We are using a 6-node HBase cluster with a Thrift Server on each of
> the
> >>> RegionServer nodes, and trying to evaluate maximum write throughput
> for our
> >>> use case (which involves many processes sending mutateRowsTs commands).
> >>> Somewhere between about 30 and 40 processes writing into the system we
> >>> cross the threshold where adding additional writers yields only very
> >>> limited returns to throughput, and I'm not sure why. We see that the
> CPU
> >>> and Disk on the DataNode/RegionServer/ThriftServer machines are not
> >>> saturated, nor is the NIC in those machines. I'm a little unsure where
> to
> >>> look next.
> >>>>
> >>>> A little more detail about our deployment:
> >>>>
> >>>> * CDH 4.1.2
> >>>> * DataNode/RegionServer/ThriftServer class: EC2 m1.xlarge
> >>>> ** RegionServer: 8GB heap
> >>>> ** ThriftServer: 1GB heap
> >>>> ** DataNode: 4GB heap
> >>>> ** EC2 ephemeral (i.e. local, not EBS) volumes used for HDFS
> >>>>
> >>>> If there's any other information that I can provide, or any other
> >>> configuration or system settings I should look at, I'd appreciate the
> >>> pointers.
> >>>>
> >>>> Thanks,
> >>>> - Dan
> >>>
> >>
> >>
>
>

I think there is a also a parameter for controlling the queue side. If you
are opening persistent connections (connections that never close), you
should probably set the queue size to 0. Because those connections will
anyways never get threads to serve them since the connections that go
through first, will hog the thread pool.

If you see "queue is full" errors in the thrift log, then there is a
bottleneck on thrift side in terms of number of workers. Of course the
assumption is that you have solved the basic problem of distributing the
load across region servers.

On Fri, Mar 1, 2013 at 10:52 AM, Ted Yu <[EMAIL PROTECTED]> wrote:

> Here're the parameters you should look at:
>
>
> hbase-server/src/main/java/org/apache/hadoop/hbase/thrift/HThreadedSelectorServerArgs.java:
>      "hbase.thrift.selector.threads";
>
> hbase-server/src/main/java/org/apache/hadoop/hbase/thrift/HThreadedSelectorServerArgs.java:
>      "hbase.thrift.worker.threads";
>
> hbase-server/src/main/java/org/apache/hadoop/hbase/thrift/TBoundedThreadPoolServer.java:
>      "hbase.thrift.threadKeepAliveTimeSec";
>
> Often times, source code is the best help :-)
>
> On Fri, Mar 1, 2013 at 10:49 AM, Dan Crosta <[EMAIL PROTECTED]> wrote:
>
> > We are generating the load from multiple machines, yes.
> >
> > Do you happen to know what the name of the setting for the number of
> > ThriftServer threads is called? I can't find anything that is obviously
> > about that in the CDH manager.
> >
> > - Dan
> >
> >
> > On Mar 1, 2013, at 1:46 PM, Varun Sharma wrote:
> >
> > > Did you try running 30-40 proc(s) on one machine and another 30-40
> > proc(s)
> > > on another machine to see if that doubles the throughput ?
> > >
> > > On Fri, Mar 1, 2013 at 10:46 AM, Varun Sharma <[EMAIL PROTECTED]>
> > wrote:
> > >
> > >> Hi,
> > >>
> > >> I don't know how many worker threads you have at the thrift servers.
> > Each
> > >> thread gets dedicated to a single connection and only serves that
> > >> connection. New connections get queued. Also, are you sure that you
> are
> > not
> > >> saturating the client side making the calls ?
> > >>
> > >> Varun
> > >>
> > >>
> > >> On Fri, Mar 1, 2013 at 9:33 AM, Jean-Daniel Cryans <
> [EMAIL PROTECTED]
> > >wrote:
> > >>
> > >>> The primary unit of load distribution in HBase is the region, make
> > >>> sure you have more than one. This is well documented in the manual
> > >>> http://hbase.apache.org/book/perf.writing.html
> > >>>
> > >>> J-D
> > >>>
> > >>> On Fri, Mar 1, 2013 at 4:17 AM, Dan Crosta <[EMAIL PROTECTED]> wrote:
> > >>>> We are using a 6-node HBase cluster with a Thrift Server on each of
> > the
> > >>> RegionServer nodes, and trying to evaluate maximum write throughput
> > for our
> > >>> use case (which involves many processes sending mutateRowsTs
> commands).
> > >>> Somewhere between about 30 and 40 processes writing into the system
> we
> > >>> cross the threshold where adding additional writers yields only very
> > >>> limited returns to throughput, and I'm not sure why. We see that the
> > CPU
> > >>> and Disk on the DataNode/RegionServer/ThriftServer machines are not
> > >>> saturated, nor is the NIC in those machines. I'm a little unsure
> where
> > to
> > >>> look next.
> > >>>>
> > >>>> A little more detail about our deployment:
> > >>>>
> > >>>> * CDH 4.1.2
> > >>>> * DataNode/RegionServer/ThriftServer class: EC2 m1.xlarge
> > >>>> ** RegionServer: 8GB heap
> > >>>> ** ThriftServer: 1GB heap
> > >>>> ** DataNode: 4GB heap
> > >>>> ** EC2 ephemeral (i.e. local, not EBS) volumes used for HDFS
> > >>>>
> > >>>> If there's any other information that I can provide, or any other
> > >>> configuration or system settings I should look at, I'd appreciate the
> > >>> pointers.
> > >>>>
> > >>>> Thanks,
> > >>>> - Dan
> > >>>
> > >>
> > >>
> >
> >
>

What performance profile do you expect?
Where does it top out (i.e. how many ops/sec)?

Also note that each data item is replicated to three nodes (by HDFS). So in a 6 machine cluster each machine would get 50% of the writes.
If you are looking for performance you really need a larger cluster to amortize this replication cost across more machines.

The other issue to watch out for is whether your keys are generated such that a single regionserver is hot spotted (you can look at the operation count on the master page).

-- Lars

________________________________
 From: Dan Crosta <[EMAIL PROTECTED]>
To: "[EMAIL PROTECTED]" <[EMAIL PROTECTED]>
Sent: Friday, March 1, 2013 4:17 AM
Subject: HBase Thrift inserts bottlenecked somewhere -- but where?
 
We are using a 6-node HBase cluster with a Thrift Server on each of the RegionServer nodes, and trying to evaluate maximum write throughput for our use case (which involves many processes sending mutateRowsTs commands). Somewhere between about 30 and 40 processes writing into the system we cross the threshold where adding additional writers yields only very limited returns to throughput, and I'm not sure why. We see that the CPU and Disk on the DataNode/RegionServer/ThriftServer machines are not saturated, nor is the NIC in those machines. I'm a little unsure where to look next.

A little more detail about our deployment:

* CDH 4.1.2
* DataNode/RegionServer/ThriftServer class: EC2 m1.xlarge
** RegionServer: 8GB heap
** ThriftServer: 1GB heap
** DataNode: 4GB heap
** EC2 ephemeral (i.e. local, not EBS) volumes used for HDFS

If there's any other information that I can provide, or any other configuration or system settings I should look at, I'd appreciate the pointers.

Thanks,
- Dan

bq. there is a also a parameter for controlling the queue side.

I guess Varun meant 'queue size'.
On http://hbase.apache.org/book.html, if you search for
'hbase.thrift.minWorkerThreads',
you would see 3 parameters. The last is hbase.thrift.maxQueuedRequests

Cheers

On Fri, Mar 1, 2013 at 11:01 AM, Varun Sharma <[EMAIL PROTECTED]> wrote:

> I think there is a also a parameter for controlling the queue side. If you
> are opening persistent connections (connections that never close), you
> should probably set the queue size to 0. Because those connections will
> anyways never get threads to serve them since the connections that go
> through first, will hog the thread pool.
>
> If you see "queue is full" errors in the thrift log, then there is a
> bottleneck on thrift side in terms of number of workers. Of course the
> assumption is that you have solved the basic problem of distributing the
> load across region servers.
>
> On Fri, Mar 1, 2013 at 10:52 AM, Ted Yu <[EMAIL PROTECTED]> wrote:
>
> > Here're the parameters you should look at:
> >
> >
> >
> hbase-server/src/main/java/org/apache/hadoop/hbase/thrift/HThreadedSelectorServerArgs.java:
> >      "hbase.thrift.selector.threads";
> >
> >
> hbase-server/src/main/java/org/apache/hadoop/hbase/thrift/HThreadedSelectorServerArgs.java:
> >      "hbase.thrift.worker.threads";
> >
> >
> hbase-server/src/main/java/org/apache/hadoop/hbase/thrift/TBoundedThreadPoolServer.java:
> >      "hbase.thrift.threadKeepAliveTimeSec";
> >
> > Often times, source code is the best help :-)
> >
> > On Fri, Mar 1, 2013 at 10:49 AM, Dan Crosta <[EMAIL PROTECTED]> wrote:
> >
> > > We are generating the load from multiple machines, yes.
> > >
> > > Do you happen to know what the name of the setting for the number of
> > > ThriftServer threads is called? I can't find anything that is obviously
> > > about that in the CDH manager.
> > >
> > > - Dan
> > >
> > >
> > > On Mar 1, 2013, at 1:46 PM, Varun Sharma wrote:
> > >
> > > > Did you try running 30-40 proc(s) on one machine and another 30-40
> > > proc(s)
> > > > on another machine to see if that doubles the throughput ?
> > > >
> > > > On Fri, Mar 1, 2013 at 10:46 AM, Varun Sharma <[EMAIL PROTECTED]>
> > > wrote:
> > > >
> > > >> Hi,
> > > >>
> > > >> I don't know how many worker threads you have at the thrift servers.
> > > Each
> > > >> thread gets dedicated to a single connection and only serves that
> > > >> connection. New connections get queued. Also, are you sure that you
> > are
> > > not
> > > >> saturating the client side making the calls ?
> > > >>
> > > >> Varun
> > > >>
> > > >>
> > > >> On Fri, Mar 1, 2013 at 9:33 AM, Jean-Daniel Cryans <
> > [EMAIL PROTECTED]
> > > >wrote:
> > > >>
> > > >>> The primary unit of load distribution in HBase is the region, make
> > > >>> sure you have more than one. This is well documented in the manual
> > > >>> http://hbase.apache.org/book/perf.writing.html
> > > >>>
> > > >>> J-D
> > > >>>
> > > >>> On Fri, Mar 1, 2013 at 4:17 AM, Dan Crosta <[EMAIL PROTECTED]>
> wrote:
> > > >>>> We are using a 6-node HBase cluster with a Thrift Server on each
> of
> > > the
> > > >>> RegionServer nodes, and trying to evaluate maximum write throughput
> > > for our
> > > >>> use case (which involves many processes sending mutateRowsTs
> > commands).
> > > >>> Somewhere between about 30 and 40 processes writing into the system
> > we
> > > >>> cross the threshold where adding additional writers yields only
> very
> > > >>> limited returns to throughput, and I'm not sure why. We see that
> the
> > > CPU
> > > >>> and Disk on the DataNode/RegionServer/ThriftServer machines are not
> > > >>> saturated, nor is the NIC in those machines. I'm a little unsure
> > where
> > > to
> > > >>> look next.
> > > >>>>
> > > >>>> A little more detail about our deployment:
> > > >>>>
> > > >>>> * CDH 4.1.2
> > > >>>> * DataNode/RegionServer/ThriftServer class: EC2 m1.xlarge
> > > >>>> ** RegionServer: 8GB heap
> > > >>>> ** ThriftServer: 1GB heap

On Mar 1, 2013, at 10:42 PM, lars hofhansl wrote:
> What performance profile do you expect?

That's a good question. Our configuration is actually already exceeding our minimum and desired performance thresholds, so I'm not too worried about it. My concern is more that I develop an understanding of where the bottlenecks are (e.g. it doesn't appear to be disk, CPU, or network bound at the moment), and develop an intuition for working with HBase in case we are ever under the gun.
> Where does it top out (i.e. how many ops/sec)?

We're doing about 20,000 writes per second sustained across 4 tables and 6 CFs. Does this sound ballpark right for 6x EC2 m1.xlarges?
> Also note that each data item is replicated to three nodes (by HDFS). So in a 6 machine cluster each machine would get 50% of the writes.
> If you are looking for performance you really need a larger cluster to amortize this replication cost across more machines.

That's only true from the HDFS perspective, right? Any given region is "owned" by 1 of the 6 regionservers at any given time, and writes are buffered to memory before being persisted to HDFS, right?

In any event, there doesn't seem to be any disk contention to speak of -- we average around 10% disk utilization at this level of load (each machine has 4 spindles of local storage, we are not using EBS).

One setting no one has mentioned yet is the DataNode handler count (dfs.datanode.handler.count) -- which is currently set to its default of 3. Should we experiment with raising that?
> The other issue to watch out for is whether your keys are generated such that a single regionserver is hot spotted (you can look at the operation count on the master page).

All of our keys are hashes or UUIDs, so the key distribution is very smooth, and this is confirmed by the "Region Servers" table on the master node's web UI.
Thanks,
- Dan

On Mar 1, 2013, at 10:53 PM, Ted Yu wrote:
> bq. there is a also a parameter for controlling the queue side.
>
> I guess Varun meant 'queue size'.
> On http://hbase.apache.org/book.html, if you search for
> 'hbase.thrift.minWorkerThreads',
> you would see 3 parameters. The last is hbase.thrift.maxQueuedRequests

I don't see that setting in CDH manager, but the book says the default is 16. If that's true in CDH, that would mean we have 6*16 == 96 which is more than the number of connections we've had at any one point (our max was 72, and we saw decreasing marginal returns from adding more workers as early as about 16 to 20).

Thanks,
- Dan

"That's only true from the HDFS perspective, right? Any given region is
"owned" by 1 of the 6 regionservers at any given time, and writes are
buffered to memory before being persisted to HDFS, right?"

Only if you disabled the WAL, otherwise each change is written to the WAL first, and then committed to the memstore.
So in the sense it's even worse. Each edit is written twice to the FS, replicated 3 times, and all that only 6 data nodes.

20k writes does seem a bit low.
-- Lars

________________________________
 From: Dan Crosta <[EMAIL PROTECTED]>
To: "[EMAIL PROTECTED]" <[EMAIL PROTECTED]>
Sent: Saturday, March 2, 2013 9:12 AM
Subject: Re: HBase Thrift inserts bottlenecked somewhere -- but where?
 
On Mar 1, 2013, at 10:42 PM, lars hofhansl wrote:
> What performance profile do you expect?

That's a good question. Our configuration is actually already exceeding our minimum and desired performance thresholds, so I'm not too worried about it. My concern is more that I develop an understanding of where the bottlenecks are (e.g. it doesn't appear to be disk, CPU, or network bound at the moment), and develop an intuition for working with HBase in case we are ever under the gun.
> Where does it top out (i.e. how many ops/sec)?

We're doing about 20,000 writes per second sustained across 4 tables and 6 CFs. Does this sound ballpark right for 6x EC2 m1.xlarges?
> Also note that each data item is replicated to three nodes (by HDFS). So in a 6 machine cluster each machine would get 50% of the writes.
> If you are looking for performance you really need a larger cluster to amortize this replication cost across more machines.

That's only true from the HDFS perspective, right? Any given region is "owned" by 1 of the 6 regionservers at any given time, and writes are buffered to memory before being persisted to HDFS, right?

In any event, there doesn't seem to be any disk contention to speak of -- we average around 10% disk utilization at this level of load (each machine has 4 spindles of local storage, we are not using EBS).

One setting no one has mentioned yet is the DataNode handler count (dfs.datanode.handler.count) -- which is currently set to its default of 3. Should we experiment with raising that?
> The other issue to watch out for is whether your keys are generated such that a single regionserver is hot spotted (you can look at the operation count on the master page).

All of our keys are hashes or UUIDs, so the key distribution is very smooth, and this is confirmed by the "Region Servers" table on the master node's web UI.
Thanks,
- Dan

On Mar 2, 2013, at 12:38 PM, lars hofhansl wrote:
> "That's only true from the HDFS perspective, right? Any given region is
> "owned" by 1 of the 6 regionservers at any given time, and writes are
> buffered to memory before being persisted to HDFS, right?"
>
> Only if you disabled the WAL, otherwise each change is written to the WAL first, and then committed to the memstore.
> So in the sense it's even worse. Each edit is written twice to the FS, replicated 3 times, and all that only 6 data nodes.

Are these writes synchronized somehow? Could there be a locking problem somewhere that wouldn't show up as utilization of disk or cpu?

What is the upshot of disabling WAL -- I assume it means that if a RegionServer crashes, you lose any writes that it has in memory but not committed to HFiles?
> 20k writes does seem a bit low.

I adjusted dfs.datanode.handler.count from 3 to 10 and now we're up to about 22-23k writes per second, but still no apparent contention for any of the basic system resources.

Any other suggestions on things to try?

Thanks,
- Dan

Make sure you are not sending a lot of put of the same rowkey. This can
cause contention in the region server side. We fixed that in our project by
aggregating all the columns for the same rowkey into the same Put object
thus when sending List of Put we made sure each Put has a unique rowkey.

On Saturday, March 2, 2013, Dan Crosta wrote:

> On Mar 2, 2013, at 12:38 PM, lars hofhansl wrote:
> > "That's only true from the HDFS perspective, right? Any given region is
> > "owned" by 1 of the 6 regionservers at any given time, and writes are
> > buffered to memory before being persisted to HDFS, right?"
> >
> > Only if you disabled the WAL, otherwise each change is written to the
> WAL first, and then committed to the memstore.
> > So in the sense it's even worse. Each edit is written twice to the FS,
> replicated 3 times, and all that only 6 data nodes.
>
> Are these writes synchronized somehow? Could there be a locking problem
> somewhere that wouldn't show up as utilization of disk or cpu?
>
> What is the upshot of disabling WAL -- I assume it means that if a
> RegionServer crashes, you lose any writes that it has in memory but not
> committed to HFiles?
>
>
> > 20k writes does seem a bit low.
>
> I adjusted dfs.datanode.handler.count from 3 to 10 and now we're up to
> about 22-23k writes per second, but still no apparent contention for any of
> the basic system resources.
>
> Any other suggestions on things to try?
>
> Thanks,
> - Dan

Asaf made a good point. See this JIRA where Nick did similar optimization:

HBASE-7747 Import tools should use a combiner to merge Puts

Cheers

On Sat, Mar 2, 2013 at 11:56 AM, Asaf Mesika <[EMAIL PROTECTED]> wrote:

> Make sure you are not sending a lot of put of the same rowkey. This can
> cause contention in the region server side. We fixed that in our project by
> aggregating all the columns for the same rowkey into the same Put object
> thus when sending List of Put we made sure each Put has a unique rowkey.
>
> On Saturday, March 2, 2013, Dan Crosta wrote:
>
> > On Mar 2, 2013, at 12:38 PM, lars hofhansl wrote:
> > > "That's only true from the HDFS perspective, right? Any given region is
> > > "owned" by 1 of the 6 regionservers at any given time, and writes are
> > > buffered to memory before being persisted to HDFS, right?"
> > >
> > > Only if you disabled the WAL, otherwise each change is written to the
> > WAL first, and then committed to the memstore.
> > > So in the sense it's even worse. Each edit is written twice to the FS,
> > replicated 3 times, and all that only 6 data nodes.
> >
> > Are these writes synchronized somehow? Could there be a locking problem
> > somewhere that wouldn't show up as utilization of disk or cpu?
> >
> > What is the upshot of disabling WAL -- I assume it means that if a
> > RegionServer crashes, you lose any writes that it has in memory but not
> > committed to HFiles?
> >
> >
> > > 20k writes does seem a bit low.
> >
> > I adjusted dfs.datanode.handler.count from 3 to 10 and now we're up to
> > about 22-23k writes per second, but still no apparent contention for any
> of
> > the basic system resources.
> >
> > Any other suggestions on things to try?
> >
> > Thanks,
> > - Dan
>

Dan already established that that is the case.

________________________________
 From: Ted Yu <[EMAIL PROTECTED]>
To: [EMAIL PROTECTED]
Sent: Saturday, March 2, 2013 12:02 PM
Subject: Re: HBase Thrift inserts bottlenecked somewhere -- but where?
 
Asaf made a good point. See this JIRA where Nick did similar optimization:

HBASE-7747 Import tools should use a combiner to merge Puts

Cheers

On Sat, Mar 2, 2013 at 11:56 AM, Asaf Mesika <[EMAIL PROTECTED]> wrote:

> Make sure you are not sending a lot of put of the same rowkey. This can
> cause contention in the region server side. We fixed that in our project by
> aggregating all the columns for the same rowkey into the same Put object
> thus when sending List of Put we made sure each Put has a unique rowkey.
>
> On Saturday, March 2, 2013, Dan Crosta wrote:
>
> > On Mar 2, 2013, at 12:38 PM, lars hofhansl wrote:
> > > "That's only true from the HDFS perspective, right? Any given region is
> > > "owned" by 1 of the 6 regionservers at any given time, and writes are
> > > buffered to memory before being persisted to HDFS, right?"
> > >
> > > Only if you disabled the WAL, otherwise each change is written to the
> > WAL first, and then committed to the memstore.
> > > So in the sense it's even worse. Each edit is written twice to the FS,
> > replicated 3 times, and all that only 6 data nodes.
> >
> > Are these writes synchronized somehow? Could there be a locking problem
> > somewhere that wouldn't show up as utilization of disk or cpu?
> >
> > What is the upshot of disabling WAL -- I assume it means that if a
> > RegionServer crashes, you lose any writes that it has in memory but not
> > committed to HFiles?
> >
> >
> > > 20k writes does seem a bit low.
> >
> > I adjusted dfs.datanode.handler.count from 3 to 10 and now we're up to
> > about 22-23k writes per second, but still no apparent contention for any
> of
> > the basic system resources.
> >
> > Any other suggestions on things to try?
> >
> > Thanks,
> > - Dan
>

Arrghhh. I meant *not* the case.

________________________________
 From: lars hofhansl <[EMAIL PROTECTED]>
To: "[EMAIL PROTECTED]" <[EMAIL PROTECTED]>
Sent: Saturday, March 2, 2013 12:50 PM
Subject: Re: HBase Thrift inserts bottlenecked somewhere -- but where?
 

Dan already established that that is the case.

________________________________
 From: Ted Yu <[EMAIL PROTECTED]>
To: [EMAIL PROTECTED]
Sent: Saturday, March 2, 2013 12:02 PM
Subject: Re: HBase Thrift inserts bottlenecked somewhere -- but where?
 
Asaf made a good point. See this JIRA where Nick did similar optimization:

HBASE-7747 Import tools should use a combiner to merge Puts

Cheers

On Sat, Mar 2, 2013 at 11:56 AM, Asaf Mesika <[EMAIL PROTECTED]> wrote:

> Make sure you are not sending a lot of put of the same rowkey. This can
> cause contention in the region server side. We fixed that in our project by
> aggregating all the columns for the same rowkey into the same Put object
> thus when sending List of Put we made sure each Put has a unique rowkey.
>
> On Saturday, March 2, 2013, Dan Crosta wrote:
>
> > On Mar 2, 2013, at 12:38 PM, lars hofhansl wrote:
> > > "That's only true from the HDFS perspective, right? Any given region is
> > > "owned" by 1 of the 6 regionservers at any given time, and writes
 are
> > > buffered to memory before being persisted to HDFS, right?"
> > >
> > > Only if you disabled the WAL, otherwise each change is written to the
> > WAL first, and then committed to the memstore.
> > > So in the sense it's even worse. Each edit is written twice to the FS,
> > replicated 3 times, and all that only 6 data nodes.
> >
> > Are these writes synchronized somehow? Could there be a locking problem
> > somewhere that wouldn't show up as utilization of disk or cpu?
> >
> > What is the upshot of disabling WAL -- I assume it means that if a
> > RegionServer crashes, you lose any writes that it has in memory but not
> > committed to HFiles?
> >
> >
> > > 20k writes does seem a bit low.
> >
> > I adjusted dfs.datanode.handler.count from 3 to 10 and now we're up to
>
 > about 22-23k writes per second, but still no apparent contention for any
> of
> > the basic system resources.
> >
> > Any other suggestions on things to try?
> >
> > Thanks,
> > - Dan
>

They are flushed to 3 nodes (but not sync'ed to disk on those replicas), so you'll eat 3 network RTTs.

I wrote a bit about this here: http://hadoop-hbase.blogspot.com/2012/05/hbase-hdfs-and-durable-sync.html

You can switch a column family to deferred log flush. In that case the edit is flushed to the 3 replication asynchronously with 1 or 2 secs.
(And if even get to finish HBASE-7801, one can control this per mutation).
-- Lars

________________________________
 From: Dan Crosta <[EMAIL PROTECTED]>
To: "[EMAIL PROTECTED]" <[EMAIL PROTECTED]>
Sent: Saturday, March 2, 2013 10:47 AM
Subject: Re: HBase Thrift inserts bottlenecked somewhere -- but where?
 
On Mar 2, 2013, at 12:38 PM, lars hofhansl wrote:
> "That's only true from the HDFS perspective, right? Any given region is
> "owned" by 1 of the 6 regionservers at any given time, and writes are
> buffered to memory before being persisted to HDFS, right?"
>
> Only if you disabled the WAL, otherwise each change is written to the WAL first, and then committed to the memstore.
> So in the sense it's even worse. Each edit is written twice to the FS, replicated 3 times, and all that only 6 data nodes.

Are these writes synchronized somehow? Could there be a locking problem somewhere that wouldn't show up as utilization of disk or cpu?

What is the upshot of disabling WAL -- I assume it means that if a RegionServer crashes, you lose any writes that it has in memory but not committed to HFiles?
> 20k writes does seem a bit low.

I adjusted dfs.datanode.handler.count from 3 to 10 and now we're up to about 22-23k writes per second, but still no apparent contention for any of the basic system resources.

Any other suggestions on things to try?

Thanks,
- Dan

Hm. This could be part of the problem in our case. Unfortunately we don't have very good control over which rowkeys will come from which workers (we're not using map-reduce or anything like it where we have that sort of control, at least not without some changes). But this is valuable information for future developments, thanks for mentioning it.

On Mar 2, 2013, at 2:56 PM, Asaf Mesika wrote:

> Make sure you are not sending a lot of put of the same rowkey. This can
> cause contention in the region server side. We fixed that in our project by
> aggregating all the columns for the same rowkey into the same Put object
> thus when sending List of Put we made sure each Put has a unique rowkey.
>
> On Saturday, March 2, 2013, Dan Crosta wrote:
>
>> On Mar 2, 2013, at 12:38 PM, lars hofhansl wrote:
>>> "That's only true from the HDFS perspective, right? Any given region is
>>> "owned" by 1 of the 6 regionservers at any given time, and writes are
>>> buffered to memory before being persisted to HDFS, right?"
>>>
>>> Only if you disabled the WAL, otherwise each change is written to the
>> WAL first, and then committed to the memstore.
>>> So in the sense it's even worse. Each edit is written twice to the FS,
>> replicated 3 times, and all that only 6 data nodes.
>>
>> Are these writes synchronized somehow? Could there be a locking problem
>> somewhere that wouldn't show up as utilization of disk or cpu?
>>
>> What is the upshot of disabling WAL -- I assume it means that if a
>> RegionServer crashes, you lose any writes that it has in memory but not
>> committed to HFiles?
>>
>>
>>> 20k writes does seem a bit low.
>>
>> I adjusted dfs.datanode.handler.count from 3 to 10 and now we're up to
>> about 22-23k writes per second, but still no apparent contention for any of
>> the basic system resources.
>>
>> Any other suggestions on things to try?
>>
>> Thanks,
>> - Dan

What is the size of your writes ?

On Sat, Mar 2, 2013 at 2:29 PM, Dan Crosta <[EMAIL PROTECTED]> wrote:

> Hm. This could be part of the problem in our case. Unfortunately we don't
> have very good control over which rowkeys will come from which workers
> (we're not using map-reduce or anything like it where we have that sort of
> control, at least not without some changes). But this is valuable
> information for future developments, thanks for mentioning it.
>
> On Mar 2, 2013, at 2:56 PM, Asaf Mesika wrote:
>
> > Make sure you are not sending a lot of put of the same rowkey. This can
> > cause contention in the region server side. We fixed that in our project
> by
> > aggregating all the columns for the same rowkey into the same Put object
> > thus when sending List of Put we made sure each Put has a unique rowkey.
> >
> > On Saturday, March 2, 2013, Dan Crosta wrote:
> >
> >> On Mar 2, 2013, at 12:38 PM, lars hofhansl wrote:
> >>> "That's only true from the HDFS perspective, right? Any given region is
> >>> "owned" by 1 of the 6 regionservers at any given time, and writes are
> >>> buffered to memory before being persisted to HDFS, right?"
> >>>
> >>> Only if you disabled the WAL, otherwise each change is written to the
> >> WAL first, and then committed to the memstore.
> >>> So in the sense it's even worse. Each edit is written twice to the FS,
> >> replicated 3 times, and all that only 6 data nodes.
> >>
> >> Are these writes synchronized somehow? Could there be a locking problem
> >> somewhere that wouldn't show up as utilization of disk or cpu?
> >>
> >> What is the upshot of disabling WAL -- I assume it means that if a
> >> RegionServer crashes, you lose any writes that it has in memory but not
> >> committed to HFiles?
> >>
> >>
> >>> 20k writes does seem a bit low.
> >>
> >> I adjusted dfs.datanode.handler.count from 3 to 10 and now we're up to
> >> about 22-23k writes per second, but still no apparent contention for
> any of
> >> the basic system resources.
> >>
> >> Any other suggestions on things to try?
> >>
> >> Thanks,
> >> - Dan
>
>

Fairly small -- row keys 32-48 bytes, column keys about the same, and values 50-100 bytes (with a few outliers that probably go up to 1k).

On Mar 3, 2013, at 6:08 AM, Varun Sharma wrote:

> What is the size of your writes ?
>
> On Sat, Mar 2, 2013 at 2:29 PM, Dan Crosta <[EMAIL PROTECTED]> wrote:
>
>> Hm. This could be part of the problem in our case. Unfortunately we don't
>> have very good control over which rowkeys will come from which workers
>> (we're not using map-reduce or anything like it where we have that sort of
>> control, at least not without some changes). But this is valuable
>> information for future developments, thanks for mentioning it.
>>
>> On Mar 2, 2013, at 2:56 PM, Asaf Mesika wrote:
>>
>>> Make sure you are not sending a lot of put of the same rowkey. This can
>>> cause contention in the region server side. We fixed that in our project
>> by
>>> aggregating all the columns for the same rowkey into the same Put object
>>> thus when sending List of Put we made sure each Put has a unique rowkey.
>>>
>>> On Saturday, March 2, 2013, Dan Crosta wrote:
>>>
>>>> On Mar 2, 2013, at 12:38 PM, lars hofhansl wrote:
>>>>> "That's only true from the HDFS perspective, right? Any given region is
>>>>> "owned" by 1 of the 6 regionservers at any given time, and writes are
>>>>> buffered to memory before being persisted to HDFS, right?"
>>>>>
>>>>> Only if you disabled the WAL, otherwise each change is written to the
>>>> WAL first, and then committed to the memstore.
>>>>> So in the sense it's even worse. Each edit is written twice to the FS,
>>>> replicated 3 times, and all that only 6 data nodes.
>>>>
>>>> Are these writes synchronized somehow? Could there be a locking problem
>>>> somewhere that wouldn't show up as utilization of disk or cpu?
>>>>
>>>> What is the upshot of disabling WAL -- I assume it means that if a
>>>> RegionServer crashes, you lose any writes that it has in memory but not
>>>> committed to HFiles?
>>>>
>>>>
>>>>> 20k writes does seem a bit low.
>>>>
>>>> I adjusted dfs.datanode.handler.count from 3 to 10 and now we're up to
>>>> about 22-23k writes per second, but still no apparent contention for
>> any of
>>>> the basic system resources.
>>>>
>>>> Any other suggestions on things to try?
>>>>
>>>> Thanks,
>>>> - Dan
>>
>>

> We're doing about 20,000 writes per second sustained across 4 tables and
6 CFs. Does this sound ballpark right for 6x EC2 m1.xlarges?

c1.xlarge will provide better performance.

m1.xlarge has pretty poor IO performance overall relative to real hardware.

Well.. so does c1.xlarge, but m1.xlarge is worse yet.

AWS is a big red flag in any discussion about performance. Systems like
Hadoop, HBase, etc. are IO intensive ultimately. When AWS built DynamoDB
they created the "high IO" instance type to host it. If you ran HBase on
that it would perform pretty well too, but you'd perhaps pale at the
monthly cost of it relative to a m1.xlarge.

Other suggestions on this thread may help, but ultimately the AWS platform
is the limiting factor, unless you move to a "high IO" or "cluster compute"
type. Ironically options for server hosting like SoftLayer are far less
expensive than AWS rates at 24/7 utilization.
On Sun, Mar 3, 2013 at 1:12 AM, Dan Crosta <[EMAIL PROTECTED]> wrote:

> On Mar 1, 2013, at 10:42 PM, lars hofhansl wrote:
> > What performance profile do you expect?
>
> That's a good question. Our configuration is actually already exceeding
> our minimum and desired performance thresholds, so I'm not too worried
> about it. My concern is more that I develop an understanding of where the
> bottlenecks are (e.g. it doesn't appear to be disk, CPU, or network bound
> at the moment), and develop an intuition for working with HBase in case we
> are ever under the gun.
>
>
> > Where does it top out (i.e. how many ops/sec)?
>
> We're doing about 20,000 writes per second sustained across 4 tables and 6
> CFs. Does this sound ballpark right for 6x EC2 m1.xlarges?
>
>
> > Also note that each data item is replicated to three nodes (by HDFS). So
> in a 6 machine cluster each machine would get 50% of the writes.
> > If you are looking for performance you really need a larger cluster to
> amortize this replication cost across more machines.
>
> That's only true from the HDFS perspective, right? Any given region is
> "owned" by 1 of the 6 regionservers at any given time, and writes are
> buffered to memory before being persisted to HDFS, right?
>
> In any event, there doesn't seem to be any disk contention to speak of --
> we average around 10% disk utilization at this level of load (each machine
> has 4 spindles of local storage, we are not using EBS).
>
> One setting no one has mentioned yet is the DataNode handler count
> (dfs.datanode.handler.count) -- which is currently set to its default of 3.
> Should we experiment with raising that?
>
>
> > The other issue to watch out for is whether your keys are generated such
> that a single regionserver is hot spotted (you can look at the operation
> count on the master page).
>
> All of our keys are hashes or UUIDs, so the key distribution is very
> smooth, and this is confirmed by the "Region Servers" table on the master
> node's web UI.
>
>
> Thanks,
> - Dan
--
Best regards,

   - Andy

Problems worthy of attack prove their worth by hitting back. - Piet Hein
(via Tom White)