用Spark和JAVA从HBase中读取数据

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.hadoop.hbase.HBaseConfiguration;
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.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

public class HBaseSparkExample {
    public static void main(String[] args) {
        // 配置HBase连接
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "localhost");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set(TableInputFormat.INPUT_TABLE, "your_table_name");

        // 创建SparkSession和SparkContext
        SparkConf sparkConf = new SparkConf().setAppName("HBaseSparkExample").setMaster("local[*]");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        SparkSession spark = SparkSession.builder().config(sparkConf).getOrCreate();

        // 读取HBase数据
        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("col1"));
        scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("col2"));

        TableInputFormat.addInputScan(scan);
        Dataset<Row> hbaseDF = spark.read().format("org.apache.hadoop.hbase.mapreduce.TableInputFormat")
                .option("hbase.table.name", "your_table_name")
                .option("hbase.columns.mapping", "cf:col1, cf:col2")
                .option("hbase.row.key", "rowKey")
                .load();

        hbaseDF.show();

        // 关闭SparkSession和SparkContext
        spark.stop();
        sc.stop();
    }
}