Cloud DataProc+Spark+Cloud BigTable（Java）依赖问题及开发实现咨询

嘿，刚好我对Cloud DataProc+Spark+BigTable这套组合熟得很，给你捋清楚依赖配置和最简代码实现，Java和Python版本都安排上，解决你遇到的问题～

先搞定依赖配置（Maven为例） #

Cloud DataProc集群本身预装了不少GCP相关依赖，但本地Eclipse开发得明确引入，还要注意版本兼容——比如Spark 3.x就得配对应版本的BigTable Hadoop Connector，不然容易踩版本冲突的坑。

给你一份靠谱的pom.xml依赖片段：

<dependencies>
    <!-- Spark核心依赖，Cloud DataProc已预装，本地调试可去掉<scope>provided</scope> -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>3.3.0</version>
        <scope>provided</scope>
    </dependency>
    <!-- Spark SQL（如果要用DataFrame的话加这个） -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>3.3.0</version>
        <scope>provided</scope>
    </dependency>
    <!-- BigTable Hadoop Connector，和Spark版本对应 -->
    <dependency>
        <groupId>com.google.cloud.bigtable</groupId>
        <artifactId>bigtable-hadoop-3.x</artifactId>
        <version>2.23.0</version>
    </dependency>
    <!-- GCP认证依赖，本地开发需要，Cloud DataProc用服务账号自动授权 -->
    <dependency>
        <groupId>com.google.cloud</groupId>
        <artifactId>google-cloud-bigtable</artifactId>
        <version>2.23.0</version>
    </dependency>
</dependencies>

⚠️ 注意：标了provided的依赖，打包时不会打入JAR包，避免和Cloud DataProc集群里的预装依赖冲突；本地调试的时候可以临时删掉这个scope，或者用Spark Submit的--jars参数手动指定依赖包。

最简功能代码：读取RDD + 批量操作 #

这里用newAPIHadoopRDD实现基础读取，再加上bulkPut写入、bulkDelete删除的示例，都是你提到的方法：

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import com.google.cloud.bigtable.hbase.BigTableConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.hbase.util.Bytes;
import java.util.List;

public class BigTableSparkDemo {
    public static void main(String[] args) {
        // 1. 初始化Spark配置，本地调试用local[*]，集群运行时自动替换
        SparkConf conf = new SparkConf()
                .setAppName("BigTable-Spark-Demo")
                .setMaster("local[*]");

        // 2. 配置BigTable连接参数，替换成你的项目/实例/表名
        String projectId = "your-gcp-project-id";
        String instanceId = "your-bigtable-instance-id";
        String tableName = "your-bigtable-table-name";

        org.apache.hadoop.conf.Configuration hbaseConf = BigTableConfiguration.configure(projectId, instanceId);
        hbaseConf.set(TableInputFormat.INPUT_TABLE, tableName);

        // 3. 创建SparkContext并执行操作
        try (JavaSparkContext sc = new JavaSparkContext(conf)) {
            // 📖 从BigTable读取数据到RDD，这里只提取RowKey，你可以按需处理Result里的列族/列
            JavaRDD<ImmutableBytesWritable> rowKeyRDD = sc.newAPIHadoopRDD(
                    hbaseConf,
                    TableInputFormat.class,
                    ImmutableBytesWritable.class,
                    org.apache.hadoop.hbase.client.Result.class
            ).map(tuple -> tuple._1());

            System.out.println("读取到的RowKey数量：" + rowKeyRDD.count());

            // 📝 批量写入BigTable（bulkPut）
            JavaRDD<Put> putRDD = sc.parallelize(List.of(
                    new Put(Bytes.toBytes("row1")).addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("col1"), Bytes.toBytes("value1")),
                    new Put(Bytes.toBytes("row2")).addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("col1"), Bytes.toBytes("value2"))
            ));

            Job putJob = Job.getInstance(hbaseConf);
            putJob.setOutputFormatClass(TableOutputFormat.class);
            TableOutputFormat.setOutputTableName(putJob, tableName);

            putRDD.map(put -> new scala.Tuple2<>(new ImmutableBytesWritable(), put))
                    .saveAsNewAPIHadoopDataset(putJob.getConfiguration());

            // 🗑️ 批量删除BigTable（bulkDelete）
            JavaRDD<Delete> deleteRDD = sc.parallelize(List.of(
                    new Delete(Bytes.toBytes("row1")),
                    new Delete(Bytes.toBytes("row2"))
            ));

            Job deleteJob = Job.getInstance(hbaseConf);
            deleteJob.setOutputFormatClass(TableOutputFormat.class);
            TableOutputFormat.setOutputTableName(deleteJob, tableName);

            deleteRDD.map(delete -> new scala.Tuple2<>(new ImmutableBytesWritable(), delete))
                    .saveAsNewAPIHadoopDataset(deleteJob.getConfiguration());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

💡 本地调试注意：要设置GOOGLE_APPLICATION_CREDENTIALS环境变量，指向你的GCP服务账号密钥文件；Cloud DataProc运行时，只要集群绑定了有权限的服务账号，就不用额外配置密钥。

如果更习惯Python，这套方案也完全可行：

依赖安装 #

本地开发直接pip装就行，Cloud DataProc集群已经预装了这些依赖：

pip install pyspark google-cloud-bigtable

⚠️ 注意：Spark版本要和Cloud DataProc集群的版本一致（比如3.3.x），避免兼容性问题。

最简功能代码 #

用官方的google-cloud-bigtable库实现读取RDD、批量写入和删除：

from pyspark import SparkContext, SparkConf
from google.cloud import bigtable
from google.cloud.bigtable.row_set import RowSet

def read_bigtable_to_rdd(sc, project_id, instance_id, table_name):
    # 初始化BigTable客户端
    client = bigtable.Client(project=project_id, admin=True)
    instance = client.instance(instance_id)
    table = instance.table(table_name)

    # 扫描所有行（也可以指定RowKey范围）
    row_set = RowSet()
    row_set.add_range_from_keys(b'', b'')

    # 读取数据并转为RDD
    rows = table.read_rows(row_set=row_set)
    row_data = []
    for row in rows:
        row_key = row.row_key.decode('utf-8')
        cells = {}
        for column, values in row.cells.items():
            cf, col = column.decode('utf-8').split(':')
            cells[f"{cf}:{col}"] = values[0].value.decode('utf-8')
        row_data.append((row_key, cells))
    return sc.parallelize(row_data)

def bulk_write_to_bigtable(project_id, instance_id, table_name, data):
    client = bigtable.Client(project=project_id, admin=True)
    instance = client.instance(instance_id)
    table = instance.table(table_name)

    rows_to_insert = []
    for row_key, columns in data:
        row = table.direct_row(row_key.encode('utf-8'))
        for cf_col, value in columns.items():
            cf, col = cf_col.split(':')
            row.set_cell(cf, col, value.encode('utf-8'))
        rows_to_insert.append(row)
    
    # 批量写入
    table.mutate_rows(rows_to_insert)

def bulk_delete_from_bigtable(project_id, instance_id, table_name, row_keys):
    client = bigtable.Client(project=project_id, admin=True)
    instance = client.instance(instance_id)
    table = instance.table(table_name)

    rows_to_delete = []
    for row_key in row_keys:
        row = table.direct_row(row_key.encode('utf-8'))
        row.delete()
        rows_to_delete.append(row)
    
    # 批量删除
    table.mutate_rows(rows_to_delete)

def main():
    # 初始化Spark配置
    conf = SparkConf().setAppName("BigTable-Spark-PyDemo").setMaster("local[*]")
    sc = SparkContext(conf=conf)

    # 替换成你的GCP参数
    project_id = "your-gcp-project-id"
    instance_id = "your-bigtable-instance-id"
    table_name = "your-bigtable-table-name"

    # 读取数据到RDD
    rdd = read_bigtable_to_rdd(sc, project_id, instance_id, table_name)
    print(f"读取到的行数：{rdd.count()}")
    rdd.foreach(print)

    # 批量写入示例数据
    sample_data = [
        ("row_py1", {"cf1:col1": "value_py1"}),
        ("row_py2", {"cf1:col1": "value_py2"})
    ]
    bulk_write_to_bigtable(project_id, instance_id, table_name, sample_data)

    # 批量删除示例数据
    bulk_delete_from_bigtable(project_id, instance_id, table_name, [row[0] for row in sample_data])

    sc.stop()

if __name__ == "__main__":
    main()

💡 提示：如果想简化操作，也可以用happybase库，语法更简洁；大规模并行处理时，建议把数据处理逻辑放到RDD的map/flatMap里，再批量提交到BigTable，效率更高。

1. 版本冲突：Spark版本和BigTable Connector必须对应——Spark 3.x用bigtable-hadoop-3.x，Spark 2.x用bigtable-hadoop-2.x，查GCP官方文档确认兼容版本；
2. 本地权限问题：本地运行必须设置GOOGLE_APPLICATION_CREDENTIALS环境变量，指向有BigTable读写权限的服务账号密钥；
3. 打包问题：Java项目用maven-shade-plugin打包时，要排除重复的HBase/Spark依赖；Python项目可以把依赖打包成wheel，或者用Cloud DataProc的初始化动作安装；
4. 集群权限问题：Cloud DataProc集群的服务账号必须拥有roles/bigtable.user或更高权限，否则会报权限拒绝错误。

内容的提问来源于stack exchange，提问作者VS_FF