[jira] [Updated] (HBASE-23062) Use TableInputFormat to read data from Hbase, when Scan.setCaching(size) the size is too big, some rowkeys will lost without exctpions.

["ZhanxiongWang (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/HBASE-23062?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

ZhanxiongWang updated HBASE-23062:
----------------------------------
    Description: 
I did the experiment in two ways. One way I use spark to read hbase, the other I use mapreduce to read hbase. In both cases, when I increase the Scan Caching size, some data will be lost. To be more accurately, When I set scan.setCaching(500), I can receive 7622 rows of data, but when I set scan.setCaching(50000), I can receive only 4226 rows of data. The seriousness of the problem is that the data is lost but there is no exceptions, it is difficult to find the reason.

My spark code is like this:
{code:java}
Configuration hbaseConfiguration = HBaseConfiguration.create();
hbaseConfiguration.set("hbase.zookeeper.property.clientPort", zkPort);
hbaseConfiguration.set("hbase.zookeeper.quorum", zkMaster);
hbaseConfiguration.set("zookeeper.znode.parent", zkPath);
hbaseConfiguration.set(TableInputFormat.INPUT_TABLE,hbaseTableName);
hbaseConfiguration.setLong("hbase.client.scanner.timeout.period",6000000);
hbaseConfiguration.setLong("hbase.rpc.timeout",6000000);
final Scan hbaseScan = new Scan();
hbaseScan.addFamily(familyName);
hbaseScan.setCaching(50000);//if Caching is too big, some rowkeys will lost!
for(String[] cell:cellNames){ 
String column = cell[0]; 
hbaseScan.addColumn(familyName,Bytes.toBytes(column));
}
hbaseScan.setStartRow(Bytes.toBytes(startRowkeyStr));
hbaseScan.setStopRow(Bytes.toBytes(endRowkeyStr));
try { 
ClientProtos.Scan scanProto = ProtobufUtil.toScan(hbaseScan); hbaseConfiguration.set(TableInputFormat.SCAN, Base64.encodeBytes(scanProto.toByteArray())); JavaPairRDD<ImmutableBytesWritable, Result> pairRDD = jsc.<ImmutableBytesWritable, Result, TableInputFormat>newAPIHadoopRDD( hbaseConfiguration, TableInputFormat.class, ImmutableBytesWritable.class, Result.class );
System.out.println("pairRDD.count(): " + pairRDD.count());
} 
catch (IOException e) { 
System.out.println("Scan Exception!!!!!! " + e.getMessage());
}
{code}
My mapreduce code is like this:
{code:java}
static class HbaseMapper extends TableMapper<ImmutableBytesWritable, Text> {
@Override protected void map(ImmutableBytesWritable key, Result value,Mapper.Context context) throws IOException, InterruptedException {
 for(Cell cell :value.rawCells()){ 
context.write(new ImmutableBytesWritable("A".getBytes()),new Text("max")); } }} 

public static void main(String[] args) throws Exception { 
org.apache.hadoop.conf.Configuration hbaseConfiguration = HBaseConfiguration.create(); hbaseConfiguration.set("hbase.zookeeper.property.clientPort", zkPort); hbaseConfiguration.set("hbase.zookeeper.quorum", zkMaster); hbaseConfiguration.set("zookeeper.znode.parent", zkPath); hbaseConfiguration.setLong("hbase.client.scanner.timeout.period",6000000); hbaseConfiguration.setLong("hbase.rpc.timeout",6000000);
Job job = Job.getInstance(hbaseConfiguration); job.setJarByClass(App.class);
List<Scan> list = new ArrayList<Scan>(); Scan scan = new Scan(); scan.addFamily(Bytes.toBytes(familyName)); 
scan.setCaching(50000);//if Caching is too big, some rowkeys will lost! 
for (String[] cell : cellNames) { 
String column = cell[0]; 
scan.addColumn(familyName,Bytes.toBytes(column)); 
} 
scan.setStartRow(Bytes.toBytes(startRowkeyStr)); 
scan.setStopRow(Bytes.toBytes(endRowkeyStr)); scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(hbaseTableName)); list.add(scan);
System.out.println("size: "+list.size()); 
TableMapReduceUtil.initTableMapperJob(list,HbaseMapper.class,ImmutableBytesWritable.class,Text.class, job);
job.setMapOutputKeyClass(ImmutableBytesWritable.class); job.setMapOutputValueClass(Text.class); 
job.setOutputKeyClass(ImmutableBytesWritable.class); 
job.setOutputValueClass(Text.class); 
FileOutputFormat.setOutputPath(job, new Path("maxTestOutput")); System.exit(job.waitForCompletion(true) ? 0 : 1);
}{code}
The pom.xml for mapreduce code is like this:

[^pom.xml]

 

  was:
I did the experiment in two ways. One way I use spark to read hbase, the other I use mapreduce to read hbase. In both cases, when I increase the Scan Caching size, some data will be lost. To be more accurately, When I set scan.setCaching(500), I can receive 7622 rows of data, but when I set scan.setCaching(50000), I can receive only 4226 rows of data. The seriousness of the problem is that the data is lost but there is no exceptions, it is difficult to find the reason.

 

 

My spark code is like this:
{code:java}
Configuration hbaseConfiguration = HBaseConfiguration.create();
hbaseConfiguration.set("hbase.zookeeper.property.clientPort", zkPort);
hbaseConfiguration.set("hbase.zookeeper.quorum", zkMaster);
hbaseConfiguration.set("zookeeper.znode.parent", zkPath);
hbaseConfiguration.set(TableInputFormat.INPUT_TABLE,hbaseTableName);
hbaseConfiguration.setLong("hbase.client.scanner.timeout.period",6000000);
hbaseConfiguration.setLong("hbase.rpc.timeout",6000000);
final Scan hbaseScan = new Scan();
hbaseScan.addFamily(familyName);
hbaseScan.setCaching(50000);//if Caching is too big, some rowkeys will lost!
for(String[] cell:cellNames){ 
String column = cell[0]; 
hbaseScan.addColumn(familyName,Bytes.toBytes(column));
}
hbaseScan.setStartRow(Bytes.toBytes(startRowkeyStr));
hbaseScan.setStopRow(Bytes.toBytes(endRowkeyStr));
try { 
ClientProtos.Scan scanProto = ProtobufUtil.toScan(hbaseScan); hbaseConfiguration.set(TableInputFormat.SCAN, Base64.encodeBytes(scanProto.toByteArray())); JavaPairRDD<ImmutableBytesWritable, Result> pairRDD = jsc.<ImmutableBytesWritable, Result, TableInputFormat>newAPIHadoopRDD( hbaseConfiguration, TableInputFormat.class, ImmutableBytesWritable.class, Result.class );
System.out.println("pairRDD.count(): " + pairRDD.count());
} 
catch (IOException e) { 
System.out.println("Scan Exception!!!!!! " + e.getMessage());
}
{code}
 

My mapreduce code is like this:

 
{code:java}
static class HbaseMapper extends TableMapper<ImmutableBytesWritable, Text> {
@Override protected void map(ImmutableBytesWritable key, Result value,Mapper.Context context) throws IOException, InterruptedException {
 for(Cell cell :value.rawCells()){ 
context.write(new ImmutableBytesWritable("A".getBytes()),new Text("max")); } }} 

public static void main(String[] args) throws Exception { 
org.apache.hadoop.conf.Configuration hbaseConfiguration = HBaseConfiguration.create(); hbaseConfiguration.set("hbase.zookeeper.property.clientPort", zkPort); hbaseConfiguration.set("hbase.zookeeper.quorum", zkMaster); hbaseConfiguration.set("zookeeper.znode.parent", zkPath); hbaseConfiguration.setLong("hbase.client.scanner.timeout.period",6000000); hbaseConfiguration.setLong("hbase.rpc.timeout",6000000);
Job job = Job.getInstance(hbaseConfiguration); job.setJarByClass(App.class);
List<Scan> list = new ArrayList<Scan>(); Scan scan = new Scan(); scan.addFamily(Bytes.toBytes(familyName)); 
scan.setCaching(50000);//if Caching is too big, some rowkeys will lost! 
for (String[] cell : cellNames) { 
String column = cell[0]; 
scan.addColumn(familyName,Bytes.toBytes(column)); 
} 
scan.setStartRow(Bytes.toBytes(startRowkeyStr)); 
scan.setStopRow(Bytes.toBytes(endRowkeyStr)); scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(hbaseTableName)); list.add(scan);
System.out.println("size: "+list.size()); 
TableMapReduceUtil.initTableMapperJob(list,HbaseMapper.class,ImmutableBytesWritable.class,Text.class, job);
job.setMapOutputKeyClass(ImmutableBytesWritable.class); job.setMapOutputValueClass(Text.class); 
job.setOutputKeyClass(ImmutableBytesWritable.class); 
job.setOutputValueClass(Text.class); 
FileOutputFormat.setOutputPath(job, new Path("maxTestOutput")); System.exit(job.waitForCompletion(true) ? 0 : 1);
}{code}
 

 

The pom.xml for mapreduce code is like this:

[^pom.xml]

 


> Use TableInputFormat to read data from Hbase, when Scan.setCaching(size) the size is too big, some rowkeys will lost without exctpions.
> ---------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: HBASE-23062
>                 URL: https://issues.apache.org/jira/browse/HBASE-23062
>             Project: HBase
>          Issue Type: Bug
>    Affects Versions: 0.98.6.1
>            Reporter: ZhanxiongWang
>            Priority: Major
>         Attachments: pom.xml
>
>
> I did the experiment in two ways. One way I use spark to read hbase, the other I use mapreduce to read hbase. In both cases, when I increase the Scan Caching size, some data will be lost. To be more accurately, When I set scan.setCaching(500), I can receive 7622 rows of data, but when I set scan.setCaching(50000), I can receive only 4226 rows of data. The seriousness of the problem is that the data is lost but there is no exceptions, it is difficult to find the reason.
> My spark code is like this:
> {code:java}
> Configuration hbaseConfiguration = HBaseConfiguration.create();
> hbaseConfiguration.set("hbase.zookeeper.property.clientPort", zkPort);
> hbaseConfiguration.set("hbase.zookeeper.quorum", zkMaster);
> hbaseConfiguration.set("zookeeper.znode.parent", zkPath);
> hbaseConfiguration.set(TableInputFormat.INPUT_TABLE,hbaseTableName);
> hbaseConfiguration.setLong("hbase.client.scanner.timeout.period",6000000);
> hbaseConfiguration.setLong("hbase.rpc.timeout",6000000);
> final Scan hbaseScan = new Scan();
> hbaseScan.addFamily(familyName);
> hbaseScan.setCaching(50000);//if Caching is too big, some rowkeys will lost!
> for(String[] cell:cellNames){ 
> String column = cell[0]; 
> hbaseScan.addColumn(familyName,Bytes.toBytes(column));
> }
> hbaseScan.setStartRow(Bytes.toBytes(startRowkeyStr));
> hbaseScan.setStopRow(Bytes.toBytes(endRowkeyStr));
> try { 
> ClientProtos.Scan scanProto = ProtobufUtil.toScan(hbaseScan); hbaseConfiguration.set(TableInputFormat.SCAN, Base64.encodeBytes(scanProto.toByteArray())); JavaPairRDD<ImmutableBytesWritable, Result> pairRDD = jsc.<ImmutableBytesWritable, Result, TableInputFormat>newAPIHadoopRDD( hbaseConfiguration, TableInputFormat.class, ImmutableBytesWritable.class, Result.class );
> System.out.println("pairRDD.count(): " + pairRDD.count());
> } 
> catch (IOException e) { 
> System.out.println("Scan Exception!!!!!! " + e.getMessage());
> }
> {code}
> My mapreduce code is like this:
> {code:java}
> static class HbaseMapper extends TableMapper<ImmutableBytesWritable, Text> {
> @Override protected void map(ImmutableBytesWritable key, Result value,Mapper.Context context) throws IOException, InterruptedException {
>  for(Cell cell :value.rawCells()){ 
> context.write(new ImmutableBytesWritable("A".getBytes()),new Text("max")); } }} 
> public static void main(String[] args) throws Exception { 
> org.apache.hadoop.conf.Configuration hbaseConfiguration = HBaseConfiguration.create(); hbaseConfiguration.set("hbase.zookeeper.property.clientPort", zkPort); hbaseConfiguration.set("hbase.zookeeper.quorum", zkMaster); hbaseConfiguration.set("zookeeper.znode.parent", zkPath); hbaseConfiguration.setLong("hbase.client.scanner.timeout.period",6000000); hbaseConfiguration.setLong("hbase.rpc.timeout",6000000);
> Job job = Job.getInstance(hbaseConfiguration); job.setJarByClass(App.class);
> List<Scan> list = new ArrayList<Scan>(); Scan scan = new Scan(); scan.addFamily(Bytes.toBytes(familyName)); 
> scan.setCaching(50000);//if Caching is too big, some rowkeys will lost! 
> for (String[] cell : cellNames) { 
> String column = cell[0]; 
> scan.addColumn(familyName,Bytes.toBytes(column)); 
> } 
> scan.setStartRow(Bytes.toBytes(startRowkeyStr)); 
> scan.setStopRow(Bytes.toBytes(endRowkeyStr)); scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(hbaseTableName)); list.add(scan);
> System.out.println("size: "+list.size()); 
> TableMapReduceUtil.initTableMapperJob(list,HbaseMapper.class,ImmutableBytesWritable.class,Text.class, job);
> job.setMapOutputKeyClass(ImmutableBytesWritable.class); job.setMapOutputValueClass(Text.class); 
> job.setOutputKeyClass(ImmutableBytesWritable.class); 
> job.setOutputValueClass(Text.class); 
> FileOutputFormat.setOutputPath(job, new Path("maxTestOutput")); System.exit(job.waitForCompletion(true) ? 0 : 1);
> }{code}
> The pom.xml for mapreduce code is like this:
> [^pom.xml]
>  



--
This message was sent by Atlassian Jira
(v8.3.4#803005)