package hbase;

import conf.SparkUtilJava;
import hbase.models.KeyValueData;
import hbase.utils.HBaseUtilJava;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import scala.Tuple2;

/**
 * Created by Abilash George Thomas on 25/9/17.
 */
public class HBaseReadJava {

    public static void main(String[] args) throws Exception {
        JavaSparkContext sparkContext = SparkUtilJava.getSparkContext();

        Configuration hbaseConfig = HBaseUtilJava.connection("example_table");

        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("name"));
        hbaseConfig.set(TableInputFormat.SCAN, HBaseUtilJava.convertScanToString(scan));

        JavaPairRDD<ImmutableBytesWritable, Result> inputBaseRDD = sparkContext.newAPIHadoopRDD(
                hbaseConfig,
                TableInputFormat.class,
                ImmutableBytesWritable.class, Result.class);

        JavaRDD<KeyValueData> resultRdd =  inputBaseRDD.map(new Function<Tuple2<ImmutableBytesWritable,Result>, Result>() {
            //@Override
            public Result call(Tuple2<ImmutableBytesWritable, Result> v1) throws Exception {
                return v1._2;
            }
        }).filter(new Function<Result, Boolean>() {

            //@Override
            public Boolean call(Result result) throws Exception {
                return true;
            }
        }).map(new Function<Result, KeyValueData>() {
            //@Override
            public KeyValueData call(Result result) throws Exception {
                return new KeyValueData(Bytes.toString(result.getValue("cf".getBytes(), "name".getBytes())), Bytes.toString(result.getValue("cf".getBytes(), "name".getBytes())));
            }
        });

        System.out.println("---------" + resultRdd.count());
    }

}

@@ -612,16 +612,40 @@
 uple2;%0A%0A
+import java.util.List;%0A%0A
 /**%0A * C
@@ -2016,16 +2016,24 @@
 lueData(
+%22name%22, 
 Bytes.to
@@ -2095,107 +2095,398 @@
 ()))
-, Bytes.toString(result.getValue(%22cf%22.getBytes(), %22name%22.getBytes())));%0A            %7D%0A        %7D);%0A%0A
+);%0A            %7D%0A        %7D);%0A%0A        // count%0A        System.out.println(%22---------%22 + resultRdd.count());%0A%0A        // print%0A        List%3CKeyValueData%3E kvList = resultRdd.collect();%0A%0A        for (KeyValueData keyValue: kvList)%7B%0A            System.out.println(%22-----------------%22);%0A            System.out.println(keyValue.getKey());%0A            System.out.println(keyValue.getValue());%0A    
     
@@ -2518,39 +2518,37 @@
 --------
-%22 + resultRdd.count());
+--------%22);%0A        %7D
 %0A    %7D%0A%0A

update spark hbase java