Spark ETL大数据迁移：并行化解析及40GB数据跨库方案

一、Spark ETL流程与核心思想 #

1. 核心流程 #

• 数据抽取（Extract）：从数据源（如BigQuery）读取分布式数据，Spark原生支持多数据源的并行读取能力
• 数据转换（Transform）：基于分布式数据集完成清洗、聚合、格式转换等操作，所有转换逻辑在集群节点并行执行
• 数据加载（Load）：将处理后的数据批量写入目标存储（如Amazon Keyspaces），利用分布式写入提升效率

2. 核心思想 #

• 分布式计算：将大拆分为多个分片，分配到集群多节点并行处理，规避单节点内存与性能瓶颈
• 惰性求值：所有转换操作仅记录逻辑，直到触发count/write等行动操作时才真正执行，Spark会自动优化执行计划
• 弹性分布式数据集（RDD）：Spark核心抽象，支持分区、并行计算与容错；DataFrame是结构化RDD封装，具备更高的查询优化效率
• 依赖优化：区分窄依赖（无需数据 shuffle，可流水线执行）与宽依赖（需跨节点数据传输），尽量减少shuffle操作降低开销

二、Spark数据并行化（拆分）的实现方式 #

Spark的并行化核心是数据分区，具体实现逻辑：

1. 数据源层面拆分：
   • 读取BigQuery时，Spark会根据表分区、数据大小自动拆分生成多个InputSplit，每个分片对应一个Task
   • 可通过option("numPartitions", "N")指定读取分区数，或用spark.sql.shuffle.partitions调整shuffle后的分区数（默认200）
2. 任务调度层面：
   • Driver节点将作业拆分为多个Stage（以shuffle操作为划分边界），每个Stage包含多个并行Task
   • Executor节点负责执行Task，每个Task处理一个数据分区，并行度由分区数与Executor核心数共同决定
3. 分区策略适配：
   • 默认采用哈希或范围分区；针对Keyspaces写入，可按Keyspaces主键设置分区，避免写入热点问题

三、40GB大数据分批迁移至Amazon Keyspaces的注意事项 #

结合你提供的代码，以下是关键优化点与注意事项：

1. 弃用逐条写入，采用Spark批量写入 #

你当前的逐条遍历Pandas DataFrame写入方式效率极低，且易引发Driver内存溢出。推荐使用Cassandra Spark Connector实现分布式批量写入：

# 优化后的批量写入逻辑
transformed_df.write \
  .format("org.apache.spark.sql.cassandra") \
  .options(table=table_name, keyspace=keyspace_name) \
  .mode("append") \
  .save()

提交作业时需添加Connector依赖：

spark-submit --packages com.datastax.spark:spark-cassandra-connector_2.12:3.4.0 your_script.py

2. 合理设置数据分区数 #

• 40GB数据建议设置160-320个分区（每个分区128MB-256MB），可通过df.repartition(N)调整分区数，平衡并行效率与调度开销
• 读取BigQuery时，可通过option("numPartitions", "200")直接指定初始分区数

3. 优化Keyspaces连接配置 #

通过Spark配置统一管理连接参数，避免每个Task重复创建连接：

# Keyspaces连接配置
spark.conf.set("spark.cassandra.connection.host", "cassandra.region.amazonaws.com")
spark.conf.set("spark.cassandra.connection.port", "9142")
spark.conf.set("spark.cassandra.auth.provider", "com.datastax.oss.driver.api.auth.auth.SigV4AuthProvider")
spark.conf.set("spark.cassandra.auth.aws.region", "region")
spark.conf.set("spark.cassandra.connection.ssl.enabled", "true")
spark.conf.set("spark.cassandra.connection.ssl.trustStore.path", "./AmazonRootCA1.pem")

# 批量写入性能配置
spark.conf.set("spark.cassandra.output.batch.size.rows", "1000")  # 每批次写入行数
spark.conf.set("spark.cassandra.output.concurrent.writes", "10")  # 单Executor并发写入数
spark.conf.set("spark.cassandra.output.consistency.level", "LOCAL_QUORUM")  # 按需设置一致性级别

4. 集群资源与内存配置 #

在YARN模式下合理分配资源，避免OOM：

spark = SparkSession \
  .builder \
  .master('yarn') \
  .appName('spark-bigquery-ks') \
  .config("spark.executor.memory", "8g") \
  .config("spark.executor.cores", "4") \
  .config("spark.driver.memory", "4g") \
  .getOrCreate()

禁止将Spark DataFrame转换为Pandas DataFrame（toPandas()），这会将全量数据拉到Driver节点内存，必然引发溢出。

5. 容错与重试机制 #

• 开启Spark任务重试：

spark.conf.set("spark.task.maxFailures", "3")  # 单个任务失败重试次数

• 配置Keyspaces连接超时与重试：

spark.conf.set("spark.cassandra.connection.timeout_ms", "10000")
spark.conf.set("spark.cassandra.read.timeout_ms", "20000")

6. 数据一致性与验证 #

• 迁移前后对比数据总量：通过df.count()与Keyspaces表行数验证
• 抽样验证数据正确性：随机抽取部分数据对比字段值
• 若源数据有实时更新，需采用快照或增量迁移方式保证一致性

7. 避免写入热点 #

• Amazon Keyspaces是分布式数据库，若写入主键分布不均会导致热点节点，严重影响性能
• 确保Spark分区键与Keyspaces主键/分区键一致，让数据均匀分布到集群节点

四、优化后的完整代码示例 #

from pyspark.sql import SparkSession

# 初始化SparkSession
spark = SparkSession \
  .builder \
  .master('yarn') \
  .appName('spark-bigquery-ks') \
  .config("spark.executor.memory", "8g") \
  .config("spark.executor.cores", "4") \
  .config("spark.driver.memory", "4g") \
  .config("spark.sql.shuffle.partitions", "200") \
  .config("spark.jars.packages", "com.datastax.spark:spark-cassandra-connector_2.12:3.4.0") \
  .getOrCreate()

# BigQuery凭证配置
spark.conf.set("credentialsFile", "./cred.json")

# 读取BigQuery全量数据
df = spark.read.format('bigquery') \
  .option('parentProject','project-id') \
  .option('dataset', 'mydataset') \
  .option('table', 'mytable') \
  .load()

# 自定义数据转换逻辑（示例）
transformed_df = df.select(
  "keyfamilyid", "recommendedfamilyid", "rank", "chi",
  "recommendationtype", "title", "location", "typepriority", "customerid"
)

# Keyspaces连接与写入配置
spark.conf.set("spark.cassandra.connection.host", "cassandra.region.amazonaws.com")
spark.conf.set("spark.cassandra.connection.port", "9142")
spark.conf.set("spark.cassandra.auth.provider", "com.datastax.oss.driver.api.auth.auth.SigV4AuthProvider")
spark.conf.set("spark.cassandra.auth.aws.region", "region")
spark.conf.set("spark.cassandra.connection.ssl.enabled", "true")
spark.conf.set("spark.cassandra.connection.ssl.trustStore.path", "./AmazonRootCA1.pem")

spark.conf.set("spark.cassandra.output.batch.size.rows", "1000")
spark.conf.set("spark.cassandra.output.concurrent.writes", "10")
spark.conf.set("spark.cassandra.output.consistency.level", "LOCAL_QUORUM")

# 写入Amazon Keyspaces
transformed_df.write \
  .format("org.apache.spark.sql.cassandra") \
  .options(table="your_table", keyspace="your_keyspace") \
  .mode("append") \
  .save()

# 关闭SparkSession
spark.stop()

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