如何高效将S3中表A迁移至含新增字段与类型变更的表B？

你的需求是把S3上的表A迁移到结构调整后的表B（新增两个字段、salary从int转string），用Pandas全量加载确实在数据量大时效率拉胯，下面给几个更高效的实现方案，按优先级排序：

1. 用Athena CTAS直接转换（最推荐，全程服务器端处理） #

既然你用awswrangler的dataset=True写入，说明表已经在Glue数据目录里了，直接用Athena的Create Table As Select语句就能搞定，不用把数据拉到本地，速度最快，代码量最少。

操作代码（Athena SQL）： #

CREATE TABLE your_database.b_table
WITH (
  format = 'PARQUET',
  external_location = 's3://your-bucket/path/to/b_table/', -- 表B的S3存储路径
  compression = 'gzip',
  partitioned_by = ARRAY['your_partition_col'] -- 如果表A有分区，这里填分区字段，和表A一致
) AS
SELECT
  uuid,
  source,
  job_title,
  CAST(salary AS VARCHAR) AS salary, -- 把int类型的salary转成string
  seniority,
  '' AS extra_info, -- 新增字段，默认给空字符串
  '' AS field -- 新增字段，默认给空字符串
FROM your_database.a_table;

执行完这条SQL，Glue会自动同步表B的元数据，和你用wr.s3.to_parquet创建的表结构完全一致。

2. 用AWS Wrangler分块处理（纯Python，灵活可控） #

如果不想写SQL，想用Python代码处理，可以用awswrangler的分块读取功能，避免一次性把所有数据加载到内存里，适合数据量中等的场景。

代码示例： #

import awswrangler as wr

# 获取表A的分区信息（如果有的话）
a_table_meta = wr.catalog.get_table(database="your_db", table="a_table")
partition_cols = [pk["Name"] for pk in a_table_meta.get("PartitionKeys", [])]

# 分块读取表A，处理后写入表B
for df_chunk in wr.s3.read_parquet(
    path=f's3://{cfg.AWS_BUCKET_NAME}/path/to/a_table/',
    dataset=True,
    database="your_db",
    table="a_table",
    chunksize=10_000_000,  # 按1000万行分块，可根据你的内存大小调整
):
    # 转换salary类型为string
    df_chunk["salary"] = df_chunk["salary"].astype(str)
    # 新增两个空字段
    df_chunk["extra_info"] = ""
    df_chunk["field"] = ""
    
    # 追加写入表B
    wr.s3.to_parquet(
        df_chunk,
        path=f's3://{cfg.AWS_BUCKET_NAME}/path/to/b_table/',
        dataset=True,
        database="your_db",
        table="b_table",
        partition_cols=partition_cols,
        compression='gzip',
        dtype=B_COLUMN_TYPES_NORMALISED_FIELDS,
        mode='append'
    )

首次运行前，可以先清空表B的S3路径，或者在第一次写入时用mode='overwrite'。

3. Dask+AWS Wrangler并行处理（超大数据量专用） #

如果数据量达到TB级，用Dask做并行处理可以充分利用多核CPU，比单进程分块处理效率更高。

代码示例： #

import awswrangler as wr
import dask.dataframe as dd

# 用Dask读取表A（自动分块并行）
ddf = wr.s3.read_parquet(
    path=f's3://{cfg.AWS_BUCKET_NAME}/path/to/a_table/',
    dataset=True,
    database="your_db",
    table="a_table",
    engine="dask"
)

# 数据转换操作（Dask会自动并行执行）
ddf["salary"] = ddf["salary"].astype(str)
ddf["extra_info"] = ""
ddf["field"] = ""

# 写入表B
wr.s3.to_parquet(
    ddf,
    path=f's3://{cfg.AWS_BUCKET_NAME}/path/to/b_table/',
    dataset=True,
    database="your_db",
    table="b_table",
    partition_cols=partition_cols,
    compression='gzip',
    dtype=B_COLUMN_TYPES_NORMALISED_FIELDS
)

方案对比 #

注意点 #

• 用Athena时，要确保Athena角色有读写对应S3桶的权限，同时目标路径不能有已存在的文件（否则CTAS会失败）。
• 分块处理的chunksize要根据你的可用内存调整，避免内存溢出。
• 写入表B时，dtype参数一定要传B_COLUMN_TYPES_NORMALISED_FIELDS，确保元数据类型正确。

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