SparkによるOBKV-HBaseへのアクセス

/*
* 現在のクラスタを次のように仮定します
* DataBaseName: test
* UserName：root
*/
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.client.connection.impl", "com.alipay.oceanbase.hbase.util.OHConnectionImpl")
conf.set(HBASE_OCEANBASE_FULL_USER_NAME, "root");
// ユーザーのパスワード
conf.set(HBASE_OCEANBASE_PASSWORD, "");
// ODP Addressの備考を参照
conf.set(HBASE_OCEANBASE_ODP_ADDR, "");
// OBKVのポートは3307（固定）
conf.setInt(HBASE_OCEANBASE_ODP_PORT, "3307");
// クラウドではODPモードを使用します(固定)
conf.setBoolean(HBASE_OCEANBASE_ODP_MODE, true);
// データベースの名前
conf.set(HBASE_OCEANBASE_DATABASE, "test");

<!-- OBKV-HBaseの依存関係には、hbase-clientとhadoop-commonが含まれます -->
<dependency>
  <groupId>com.oceanbase</groupId>
  <artifactId>obkv-hbase-client</artifactId>
  <version>2.2.1</version>
</dependency>

<!-- OBKV-HBaseの依存関係 -->
<dependency>
  <groupId>com.oceanbase</groupId>
  <artifactId>obkv-hbase-client</artifactId>
  <version>${hbase.client.version}</version>
  <exclusions>
    <exclusion>
      <artifactId>jackson-databind</artifactId>
      <groupId>com.fasterxml.jackson.core</groupId>
    </exclusion>
  </exclusions>
</dependency>
<!-- Sparkの依存関係 -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.12</artifactId>
  <version>${spark.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql_2.12</artifactId>
  <version>${spark.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming_2.12</artifactId>
  <version>${spark.version}</version>
</dependency>
<!-- SparkとHBaseコネクタの依存関係 -->
<dependency>
  <groupId>org.apache.hbase.connectors.spark</groupId>
  <artifactId>hbase-spark</artifactId>
  <version>${hbase.spark.version}</version>
  <exclusions>
    <exclusion>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-client</artifactId>
    </exclusion>
    <exclusion>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-common</artifactId>
    </exclusion>
  </exclusions>
</dependency>

-- テーブルグループを作成
CREATE TABLEGROUP `spark_demo` SHARDING = 'ADAPTIVE';

-- テーブルを作成
CREATE TABLE `spark_demo$cf1` (
    `K` varbinary(1024) NOT NULL,
    `Q` varbinary(256) NOT NULL,
    `T` bigint(20) NOT NULL,
    `V` varbinary(1024) DEFAULT NULL,
    PRIMARY KEY (`K`, `Q`, `T`)
) TABLEGROUP = `spark_demo` partition by key(`K`) partitions 17;
CREATE TABLE `spark_demo$cf2` (
    `K` varbinary(1024) NOT NULL,
    `Q` varbinary(256) NOT NULL,
    `T` bigint(20) NOT NULL,
    `V` varbinary(1024) DEFAULT NULL,
    PRIMARY KEY (`K`, `Q`, `T`)
) TABLEGROUP = `spark_demo` partition by key(`K`) partitions 17;

// データの書き込み
public void testSparkReadAfterWriteHBase() throws Exception {
    // SparkContextの初期化
    SparkConf sparkConf = new SparkConf().setAppName("HBaseWriteExample").setMaster("local");
    JavaSparkContext jsc = new JavaSparkContext(sparkConf);
    // 書き込み用のHBase設定
    Configuration hbaseConf = HBaseConfiguration.create();
    hbaseConf.set(TableOutputFormat.OUTPUT_TABLE, "spark_demo");
    hbaseConf.set("mapreduce.job.output.key.class", ImmutableBytesWritable.class.getName());
    hbaseConf.set("mapreduce.job.output.value.class", Put.class.getName());
    hbaseConf.set("mapreduce.outputformat.class", TableOutputFormat.class.getName());
    // RDDデータの作成(Key: RowKey, Value: Tuple of (ColumnFamily, ColumnName, Value)))
    JavaRDD<Tuple2<byte[], Tuple3<String, String, String>>> data = jsc.parallelize(Arrays.asList(
        new Tuple2<>(Bytes.toBytes(row1), new Tuple3<>(family1, "name", john_name)),
        new Tuple2<>(Bytes.toBytes(row1), new Tuple3<>(family1, "age", john_age)),
        new Tuple2<>(Bytes.toBytes(row1), new Tuple3<>(family1, "gender", john_gender)),
        new Tuple2<>(Bytes.toBytes(row2), new Tuple3<>(family1, "name", lily_name)),
        new Tuple2<>(Bytes.toBytes(row2), new Tuple3<>(family1, "age", lily_age)),
        new Tuple2<>(Bytes.toBytes(row2), new Tuple3<>(family1, "gender", lily_gender)),
        new Tuple2<>(Bytes.toBytes(row3), new Tuple3<>(family1, "name", walter_name)),
        new Tuple2<>(Bytes.toBytes(row3), new Tuple3<>(family1, "age", walter_age)),
        new Tuple2<>(Bytes.toBytes(row3), new Tuple3<>(family1, "gender", walter_gender))
    ));
    // RDDをHBase形式(ImmutableBytesWritable, Put)にマッピング
    JavaPairRDD<ImmutableBytesWritable, Put> hbaseWriteRDD = data.mapToPair(tuple -> {
        byte[] rowKey = tuple._1;
        String family = tuple._2._1();
        String qualifier = tuple._2._2();
        String value = tuple._2._3();

        Put put = new Put(rowKey);
        put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
        return new Tuple2<>(new ImmutableBytesWritable(rowKey), put);
    });
    // データをHBaseに書き込む
    hbaseWriteRDD.saveAsNewAPIHadoopDataset(hbaseConf);
    // HBaseから読み取る
    hbaseConf.set(TableInputFormat.INPUT_TABLE, "spark_demo");
    Scan scan = new Scan();
    scan.addFamily(Bytes.toBytes(family1));
    scan.addFamily(Bytes.toBytes(family2));
    scan.setCaching(100);
    String scanToString = org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil.convertScanToString(scan);
    hbaseConf.set(TableInputFormat.SCAN, scanToString);
    JavaPairRDD<ImmutableBytesWritable, Result> hbaseReadRDD = jsc.newAPIHadoopRDD(
        hbaseConf,
        TableInputFormat.class,
        ImmutableBytesWritable.class,
        Result.class
    );
    // データの処理
    hbaseReadRDD.flatMapToPair(tuple -> {
        ImmutableBytesWritable key = tuple._1();
        Result result = tuple._2();
        String rowKey = Bytes.toString(key.get());
        List<Tuple2<String, String>> res = new ArrayList<>();
        for (Cell c : result.rawCells()) {
            String familyValue = Bytes.toString(CellUtil.cloneValue(c));
            String column = Bytes.toString(CellUtil.cloneQualifier(c));
                    Tuple2<String, String> value = new Tuple2<>(rowKey, String.format("Get result from cf1: K: %s, Q: %s, V: %s", Bytes.toString(result.getRow()), column, familyValue));
                    res.add(value);
        }
                return res.iterator();
    }).collect().forEach(System.out::println);
    // SparkContextを停止
    jsc.stop();
}

public void testSparkSessionSQLHBase() throws Exception {
    SparkConf conf = new SparkConf().setAppName("SparkSQLHBaseExample").setMaster("local[*]");
    conf.set("spark.sql.defaultUrlStreamHandlerFactory.enabled", "false"); // avoid Connection Casting Exception
    JavaSparkContext jsc = new JavaSparkContext(conf);
    // Initialize SparkSession
    SparkSession spark = SparkSession.builder()
            .appName("SparkSQLHBaseExample")
            .config(conf)
            .getOrCreate();
    // HBase configuration
    Configuration hbaseConf = ObHTableTestUtil.newConfiguration();
    // Initialize HBaseContext
    String tmpHdfsConfigPath = "/your-hdfs-path/tmp/hbase-config";
    HBaseContext hbaseContext = new HBaseContext(jsc.sc(), hbaseConf, tmpHdfsConfigPath);
    String tableName = "spark_demo";
    String hbase_column_mapping = "id STRING :key, " +
            "name STRING cf1:name, " +
            "gender STRING cf1:gender, " +
            "age STRING cf1:age";
    Dataset<Row> hbaseDF = spark.read()
            .format("org.apache.hadoop.hbase.spark")
            .option("hbase.columns.mapping", hbase_column_mapping)
            // disable column pushdown optimization，OBKV-HBase does not support SparkSQLPushDownFilter yet
            .option("hbase.spark.pushdown.columnfilter", false)
            .option("hbase.table", tableName)
            .load();
    hbaseDF.printSchema();
    // select * from table
    hbaseDF.show();
    // Perform SQL queries on the DataFrame
    hbaseDF.createOrReplaceTempView(tableName);
    // select with condition
    Dataset<Row> result = spark.sql("SELECT * FROM spark_demo WHERE age > 30");
    result.show();
    jsc.stop();
    spark.stop();
}