基于Node.js+MySQL的遗留电商应用与Google BigQuery同步方案咨询

阿华AIGC实验室

2026-5-21

Hey there! 针对你的Node.js+MySQL遗留电商应用同步数据到Google BigQuery的需求，结合你能接受15-30分钟延迟、需要覆盖历史数据+新增/更新数据的目标，我整理了几个实用方案，帮你一步步搞定：

方案1：历史数据一次性导出+定时增量同步（最匹配你的需求）

这个方案不需要修改遗留应用代码，成本低，完全满足15-30分钟的延迟要求，是最适合你的选择。

第一步：导入历史数据

先把MySQL中已有的历史数据一次性同步到BigQuery，推荐用CSV导出+BigQuery批量加载的方式：

导出MySQL数据为CSV：
用mysqldump命令导出指定表（比如订单表orders）的CSV文件，注意格式兼容BigQuery：
```
mysqldump -u your-mysql-user -p --tab=/tmp \
  --fields-terminated-by=',' \
  --fields-enclosed-by='"' \
  --lines-terminated-by='\n' \
  your-ecommerce-db orders
```
也可以用MySQL的SELECT ... INTO OUTFILE语句导出，适合自定义筛选条件的场景。
上传CSV到Google Cloud Storage（GCS）：
把导出的CSV文件上传到GCS bucket（如果没有可以先创建一个），这比直接本地导入BigQuery更稳定，尤其是大数据量的情况。

加载CSV到BigQuery：
用BigQuery的命令行工具bq或者Node.js SDK执行加载作业：

bq load --source_format=CSV --skip_leading_rows=1 \
  your-gcp-project:your-bq-dataset.orders \
  gs://your-gcs-bucket/orders.txt \
  "order_id:INT64,customer_id:INT64,total_amount:FLOAT64,created_at:TIMESTAMP,updated_at:TIMESTAMP"

这里的最后一个参数是表结构定义，也可以让BigQuery自动检测模式（添加--autodetect参数）。

第二步：定时增量同步新增/更新数据

用Node.js的定时任务工具（比如node-schedule或cron），每隔15-30分钟拉取MySQL中上次同步后新增或更新的数据，批量写入BigQuery：

核心思路：

依赖MySQL表中的updated_at字段（确保这个字段会在数据新增/修改时自动更新），每次同步拉取updated_at在上次同步时间到当前时间之间的数据。
用BigQuery的批量插入接口，比流式插入更节省成本，且适合定时同步的场景。

代码示例（Node.js）：

const mysql = require('mysql2/promise');
const { BigQuery } = require('@google-cloud/bigquery');
const schedule = require('node-schedule');

// 配置信息
const mysqlConfig = {
  host: 'your-mysql-host',
  user: 'your-mysql-user',
  password: 'your-mysql-password',
  database: 'your-ecommerce-db'
};
const bigquery = new BigQuery({ projectId: 'your-gcp-project-id' });
const bqTable = bigquery.dataset('your-bq-dataset').table('orders');

// 初始化上次同步时间（设为历史数据导入完成的时间）
let lastSyncTimestamp = new Date('2024-01-01T00:00:00');

// 定时任务：每30分钟执行一次
schedule.scheduleJob('*/30 * * * *', async () => {
  const currentTimestamp = new Date();
  console.log(`Starting sync from ${lastSyncTimestamp} to ${currentTimestamp}`);

  try {
    // 1. 查询MySQL增量数据
    const mysqlConn = await mysql.createConnection(mysqlConfig);
    const [rows] = await mysqlConn.execute(
      `SELECT * FROM orders 
       WHERE updated_at > ? AND updated_at <= ?`,
      [lastSyncTimestamp, currentTimestamp]
    );
    await mysqlConn.end();

    if (rows.length === 0) {
      console.log('No new data to sync');
      lastSyncTimestamp = currentTimestamp;
      return;
    }

    // 2. 转换数据格式（适配BigQuery类型，比如日期转ISO字符串）
    const formattedRows = rows.map(row => ({
      ...row,
      created_at: row.created_at.toISOString(),
      updated_at: row.updated_at.toISOString()
    }));

    // 3. 批量写入BigQuery
    const [job] = await bqTable.insert(formattedRows);
    await job.promise(); // 等待加载作业完成

    console.log(`Successfully synced ${rows.length} rows`);
    // 更新上次同步时间
    lastSyncTimestamp = currentTimestamp;
  } catch (error) {
    console.error('Sync failed:', error);
    // 这里可以添加告警逻辑，比如发送邮件/消息通知
  }
});