import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{ConnectionFactory, Put}
import org.apache.hadoop.hbase.spark.HBaseContext
import org.apache.spark.{SparkConf, SparkContext}

val sparkConf = new SparkConf().setAppName("SparkToHBase")
val sc = new SparkContext(sparkConf)

// 配置HBase
val hbaseConf = HBaseConfiguration.create()
hbaseConf.set("hbase.zookeeper.quorum", "zk1.example.com,zk2.example.com")
hbaseConf.set("hbase.zookeeper.property.clientPort", "2181")

// 创建HBaseContext
val hbaseContext = new HBaseContext(sc, hbaseConf)

// 示例数据
val rdd = sc.parallelize(Array(
  ("row1", "cf1", "col1", "value1"),
  ("row2", "cf1", "col1", "value2")
))

// 转换为Put对象
val putRdd = rdd.map{ case (rowKey, cf, col, value) =>
  val put = new Put(rowKey.getBytes)
  put.addColumn(cf.getBytes, col.getBytes, value.getBytes)
  put
}

// 批量写入HBase
hbaseContext.bulkPut[Put](putRdd, TableName.valueOf("your_table_name"))

import org.apache.spark.sql.{SaveMode, SparkSession}

val spark = SparkSession.builder()
  .appName("SparkDFToHBase")
  .getOrCreate()

// 创建示例DataFrame
val data = Seq(
  ("row1", "value1"),
  ("row2", "value2")
)
val df = spark.createDataFrame(data).toDF("rowkey", "value")

// 配置HBase连接
df.write
  .format("org.apache.hadoop.hbase.spark")
  .option("hbase.table", "your_table_name")
  .option("hbase.columns.mapping", 
    "rowkey STRING :key, value STRING cf1:col1")
  .option("hbase.spark.use.hbasecontext", "true")
  .mode(SaveMode.Append)
  .save()