如何解决DataFrame加载到Google BigQuery后scSequence列顺序混乱问题？

问题根源 #

当前代码存在三个核心问题：

1. 依赖df.index +1生成scSequence，但DataFrame的索引可能因过滤、重排等操作不连续，导致序列本身不符合预期；
2. BigQuery加载数据时不保证保留插入顺序，即使DataFrame内序列正确，加载后行的存储顺序也可能被打乱，最终scSequence看起来混乱；
3. 原函数注释提到要基于表中最后一行的scSequence递增，但代码完全没实现这个逻辑，只是用本地索引生成值，导致多次追加数据时序列断裂。

解决方法 #

方法一：Python端修正（生成可靠的连续序列） #

通过查询目标表的最大scSequence值，以此为起点生成连续序列，同时明确写入模式确保追加逻辑正确：

from google.cloud import bigquery

def load_table_from_dataframe(df, source_type):
    """
    Loads a pandas DataFrame into the specified BigQuery table.
    Adds a new column 'scSequence' with incremented row numbers based on the last row in the table.

    Args:
        df (pandas.DataFrame): The DataFrame to be loaded.
        source_type (str): The type of source (e.g., "RX", "MS").
    """
    table_id = f"{PROJECT_ID}.{DATASET_ID}.{TABLE_MAPPING[source_type]}"
    client = bigquery.Client()

    # 获取目标表当前最大scSequence，表为空或不存在时默认0
    max_seq = 0
    try:
        query = f"SELECT IFNULL(MAX(scSequence), 0) AS max_seq FROM `{table_id}`"
        query_result = client.query(query).to_dataframe()
        max_seq = query_result['max_seq'][0]
    except Exception:
        pass  # 表不存在时直接从1开始

    # 生成连续递增的scSequence
    df.insert(0, "scSequence", range(max_seq + 1, max_seq + len(df) + 1))

    # 配置加载任务为追加模式（如需覆盖则改为WRITE_TRUNCATE）
    job_config = bigquery.LoadJobConfig(
        write_disposition=bigquery.WriteDisposition.WRITE_APPEND
    )
    job = client.load_table_from_dataframe(df, table_id, job_config=job_config)
    job.result()

    print(f"Data successfully loaded into {table_id}")

注意：如果存在多进程同时写入的场景，建议结合BigQuery的自增列（如GENERATED ALWAYS AS IDENTITY）来避免序列冲突，上述代码适用于单进程或低并发场景。

方法二：BigQuery端修正（已加载数据的序列修复） #

如果数据已经加载到BigQuery，可通过窗口函数重新生成有序的scSequence，前提是需要有一个确定顺序的业务字段（如时间戳、业务主键）：

方案1：重建表并生成序列

CREATE OR REPLACE TABLE `your-project.your-dataset.your-table` AS
SELECT 
  ROW_NUMBER() OVER (ORDER BY your_order_column) AS scSequence,
  * EXCEPT(scSequence)
FROM `your-project.your-dataset.your-table`;

将your_order_column替换为实际用来排序的字段（比如数据的创建时间create_time）。

方案2：直接更新现有表

UPDATE `your-project.your-dataset.your-table`
SET scSequence = new_seq
FROM (
  SELECT 
    scSequence AS old_seq,
    ROW_NUMBER() OVER (ORDER BY your_order_column) AS new_seq
  FROM `your-project.your-dataset.your-table`
) AS seq_mapping
WHERE `your-project.your-dataset.your-table`.scSequence = seq_mapping.old_seq;

注意：BigQuery的UPDATE操作会产生存储和计算成本，执行前需确认表的权限和数据量。

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