DataProc任务从BigQuery同步至MySQL耗时过长的优化咨询

针对100GB数据同步耗时超1天的问题，可从读取、写入、集群配置等多维度优化，以下是具体方案：

一、BigQuery读取阶段优化 #

• 调整并行读取策略：
  原代码的maxParallelism=1000可能过高，导致集群资源竞争。建议根据集群核心数调整为2-3倍的并行任务数，同时结合数据分布选择合适的分区列进行范围读取，避免数据倾斜：

  dataframe = spark.read.format("com.google.cloud.spark.bigquery") \
    .option("maxParallelism", 200)  # 按集群核心数调整，比如20个executor×4核=80，设为200
    .option("partitionColumn", "id")  # 选分布均匀的主键或时间分区列
    .option("lowerBound", 1)
    .option("upperBound", 10000000)  # 对应partitionColumn的取值范围
    .option("numPartitions", 200)  # 与maxParallelism匹配
    .load("<table>")
  

• 移除不必要的cache()：
  100GB数据缓存会占用大量内存，甚至触发磁盘溢出拖慢速度，若无需重复使用数据集，直接删除.cache()。

二、MySQL写入阶段优化 #

• 优化JDBC批量写入参数：
  增大batchsize至10000-50000（需匹配MySQL的max_allowed_packet参数，避免单批次数据超出限制），同时开启useServerPrepStmts提升批量插入效率：

  dataframe.write.format('jdbc') \
    .option('url', f'jdbc:mysql://{MYSQL_INSTANCE_CONNECTION_NAME}/{MYSQL_DATABASE_NAME}?useServerPrepStmts=true&rewriteBatchedStatements=true') \
    .option('dbtable', MYSQL_TABLE_NAME) \
    .option('user', MYSQL_USER) \
    .option('password', MYSQL_PASSWORD) \
    .option('batchsize', 50000) \
    .option('numPartitions', 200)  # 与读取阶段分区数一致，避免任务数不匹配
    .mode('append') \
    .save()
  

• 调整MySQL服务器配置：
  在MySQL端修改以下参数（需重启服务生效）：
  • innodb_buffer_pool_size：设为服务器内存的50%-70%，提升数据缓存能力
  • innodb_log_file_size：调整为1GB-4GB，减少日志刷写频率
  • innodb_flush_log_at_trx_commit：设为2，牺牲少量事务一致性换取写入性能
  • max_connections：调大至支持Spark的并行写入任务数

三、Spark集群配置优化 #

• 扩容集群资源：
  增加executor数量、内存和核心数，比如提交作业时指定：

  spark-submit \
    --num-executors 20 \
    --executor-cores 4 \
    --executor-memory 16G \
    --driver-memory 8G \
    your_script.py
  

• 避免数据倾斜：
  若源表存在热点数据，提前对数据集重分区，确保每个分区数据量均匀：

  dataframe = dataframe.repartition(200, "id")  # 按分布均匀的列重分区
  

• 关闭推测执行：
  若集群任务执行时间差异大，推测执行会重复运行任务浪费资源，在Spark配置中添加：

  spark = SparkSession.builder.appName('MySQL Data Loader') \
    .config("spark.speculation", "false") \
    .getOrCreate()
  

四、替代高效方案（若业务允许） #

用BigQuery直接导出到GCS，再通过MySQL原生工具导入，性能远优于JDBC：

1. BigQuery导出到GCS：

   from google.cloud import bigquery
   client = bigquery.Client()
   destination_uri = "gs://your-bucket/export-*.parquet"
   table_ref = client.get_table("your-project.your-dataset.your-table")
   extract_job = client.extract_table(table_ref, destination_uri, location="US")
   extract_job.result()
   

2. 导入到MySQL：
   • 若为云MySQL（如GCP Cloud SQL），直接用gcloud sql import csv命令导入
   • 自建MySQL可将GCS文件下载到服务器，用LOAD DATA INFILE批量导入

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