如何将Sparks recommendProductsForUsers保存到Hbase

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class SparkHBaseExample {
    public static void main(String[] args) {
        // 创建SparkSession
        SparkSession spark = SparkSession.builder()
                .appName("SparkHBaseExample")
                .getOrCreate();

        // 生成recommendProductsForUsers方法生成的DataFrame或Dataset
        Dataset<Row> recommendDF = spark.read().format("csv").load("recommendations.csv");

        // 将DataFrame或Dataset转换为RDD
        JavaRDD<Row> recommendRDD = recommendDF.toJavaRDD();

        // 在每个分区中将推荐的产品列表保存到HBase
        recommendRDD.foreachPartition(new VoidFunction<Iterator<Row>>() {
            @Override
            public void call(Iterator<Row> iterator) throws Exception {
                // 创建HBase连接
                org.apache.hadoop.conf.Configuration hbaseConfig = HBaseConfiguration.create();
                Connection hbaseConnection = ConnectionFactory.createConnection(hbaseConfig);
                Table hbaseTable = hbaseConnection.getTable("recommendations");

                // 遍历每个分区的数据
                while (iterator.hasNext()) {
                    Row row = iterator.next();
                    String userId = row.getString(0);
                    String recommendedProducts = row.getString(1);

                    // 创建Put对象并添加列值
                    Put put = new Put(userId.getBytes());
                    put.addColumn("cf".getBytes(), "recommendedProducts".getBytes(), recommendedProducts.getBytes());

                    // 将Put对象插入到HBase表中
                    hbaseTable.put(put);
                }

                // 关闭HBase连接
                hbaseTable.close();
                hbaseConnection.close();
            }
        });

        // 关闭SparkSession
        spark.close();
    }
}