Python Redshift UNLOAD 导出 Parquet 并写

脚本专家

在数据工程与数据分析场景中，经常需要将本地 Parquet 文件与 Amazon Redshift 中的查询结果合并，并输出为单个 Excel 文件的不同 Sheet。本文基于 Python 生态中的 pandas、redshift_connector、awswrangler 和 openpyxl，实现一套完整的从本地读取、Redshift UNLOAD 至 S3、下载合并、排序再到 Excel 多 Sheet 写入的自动化流程。

## 环境与依赖
首先安装必要的 Python 包：

2. pip install pandas pyarrow redshift_connector awswrangler openpyxl boto3

复制代码

导入时依次引入：

2. import os
3. import pandas as pd
4. import redshift_connector
5. import awswrangler as wr
6. import boto3
7. from datetime import datetime
8. from pathlib import Path

复制代码

## 1. 从本地 Parquet 读取 DataFrame（df_1）
使用 pandas 的 read_parquet 方法，引擎指定为 pyarrow（默认），对大列式存储文件性能更优。可通过 use_threads=True 启用多线程读取，memory_map=True 使用内存映射降低 I/O 开销。

2. parquet_path = "./data/source_data.parquet"
3. df_1 = pd.read_parquet(parquet_path, engine="pyarrow")

复制代码

## 2. 从 Amazon Redshift 通过 UNLOAD 导出数据至 S3 并加载为 DataFrame（df_2）
此环节整合了数据库连接、UNLOAD 操作、S3 读取三个步骤。

### 2.1 建立 Redshift 连接
使用 redshift_connector.connect 方法，从环境变量中读取主机、数据库、端口、用户名和密码：

2. conn = redshift_connector.connect(
3.     host=os.getenv("REDSHIFT_HOST"),
4.     database=os.getenv("REDSHIFT_DATABASE"),
5.     port=int(os.getenv("REDSHIFT_PORT", 5439)),
6.     user=os.getenv("REDSHIFT_USER"),
7.     password=os.getenv("REDSHIFT_PASSWORD")
8. )

复制代码

### 2.2 执行 UNLOAD 到 S3
利用 awswrangler 的 wr.redshift.unload_to_files 封装 Redshift 的 UNLOAD 命令。为避免路径覆盖，在 S3 路径中嵌入时间戳：

2. timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
3. s3_unload_path = f"s3://your-bucket/redshift_unload/{timestamp}/"
5. sql_query = """
6. SELECT id, column_a, column_b, created_at
7. FROM your_schema.your_table
8. WHERE created_at >= '2026-01-01'
9. """
11. wr.redshift.unload_to_files(
12.     sql=sql_query,
13.     path=s3_unload_path,
14.     con=conn,
15.     iam_role=os.getenv("REDSHIFT_IAM_ROLE"),
16.     unload_format="PARQUET",
17.     parallel=True,
18.     region=os.getenv("AWS_REGION", "us-east-1")
19. )

复制代码

关键参数说明：
- unload_format 选 PARQUET 可获得更高压缩率，后续读取更快。
- parallel=True 利用 Redshift 集群切片并行导出，显著提升大数据量场景下的速度。
- iam_role 必须具有 s3:PutObject 和 s3:ListBucket 权限，否则会遇到 AccessDenied。
- region 需与 S3 桶区域一致；若 Redshift 与 S3 在不同区域，务必显式传递 region 参数。

### 2.3 从 S3 读取为 DataFrame
使用 boto3.Session 配置区域，再通过 awswrangler.s3.read_parquet 自动识别 S3 路径下的所有 Parquet 文件并合并读取：

2. session = boto3.Session(region_name=os.getenv("AWS_REGION", "us-east-1"))
3. df_2 = wr.s3.read_parquet(path=s3_unload_path, boto3_session=session)

复制代码

如果 df_2 数据量极大，可使用 chunked=True 分块读取以控制内存。

### 2.4 关闭连接

2. conn.close()

复制代码

## 3. 统一按 id 列排序
为保证 Excel 中两个 Sheet 的数据顺序一致，对两个 DataFrame 均按 id 升序排列，并重置索引：

2. df_1 = df_1.sort_values(by="id", ascending=True, ignore_index=True)
3. df_2 = df_2.sort_values(by="id", ascending=True, ignore_index=True)

复制代码

ignore_index=True 避免旧索引写入 Excel。

## 4. 写入同一 Excel 文件的不同 Sheet
使用 pd.ExcelWriter 搭配 openpyxl 引擎（对 .xlsx 格式支持良好），通过 with 语句确保文件自动保存关闭：

2. output_path = Path("./output/mult_sheets_1.xlsx")
3. output_path.parent.mkdir(parents=True, exist_ok=True)
5. with pd.ExcelWriter(output_path, engine="openpyxl") as writer:
6.     df_1.to_excel(writer, sheet_name="df_1", index=False)
7.     df_2.to_excel(writer, sheet_name="df_2", index=False)

复制代码

index=False 防止行索引写入。若需更多 Sheet 继续调用 to_excel 即可。

## 5. 完整代码整合
将上述片段合并为可独立运行的脚本：

2. import os
3. import pandas as pd
4. import redshift_connector
5. import awswrangler as wr
6. import boto3
7. from datetime import datetime
8. from pathlib import Path
10. # 配置
11. parquet_path = "./data/source_data.parquet"
12. output_path = Path("./output/mult_sheets_1.xlsx")
13. output_path.parent.mkdir(parents=True, exist_ok=True)
15. sql_query = """
16. SELECT id, column_a, column_b, created_at
17. FROM your_schema.your_table
18. WHERE created_at >= '2026-01-01'
19. """
21. # 1. 读取本地 Parquet
22. df_1 = pd.read_parquet(parquet_path, engine="pyarrow")
24. # 2. 连接 Redshift
25. conn = redshift_connector.connect(
26.     host=os.getenv("REDSHIFT_HOST"),
27.     database=os.getenv("REDSHIFT_DATABASE"),
28.     port=int(os.getenv("REDSHIFT_PORT", 5439)),
29.     user=os.getenv("REDSHIFT_USER"),
30.     password=os.getenv("REDSHIFT_PASSWORD")
31. )
33. # 3. UNLOAD 到 S3
34. timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
35. s3_unload_path = f"s3://your-bucket/redshift_unload/{timestamp}/"
36. wr.redshift.unload_to_files(
37.     sql=sql_query,
38.     path=s3_unload_path,
39.     con=conn,
40.     iam_role=os.getenv("REDSHIFT_IAM_ROLE"),
41.     unload_format="PARQUET",
42.     parallel=True,
43.     region=os.getenv("AWS_REGION", "us-east-1")
44. )
46. # 4. 从 S3 读取
47. df_2 = wr.s3.read_parquet(path=s3_unload_path, boto3_session=boto3.Session(region_name=os.getenv("AWS_REGION", "us-east-1")))
49. # 5. 关闭连接
50. conn.close()
52. # 6. 排序
53. df_1 = df_1.sort_values(by="id", ascending=True, ignore_index=True)
54. df_2 = df_2.sort_values(by="id", ascending=True, ignore_index=True)
56. # 7. 写入 Excel
57. with pd.ExcelWriter(output_path, engine="openpyxl") as writer:
58.     df_1.to_excel(writer, sheet_name="df_1", index=False)
59.     df_2.to_excel(writer, sheet_name="df_2", index=False)
60. print(f"Excel 已生成：{output_path}")

复制代码

## 6. 常见问题与调试建议
### 6.1 UNLOAD 权限错误
若遇到 AccessDenied，需检查 IAM 角色的 s3:PutObject 和 s3:ListBucket 权限是否授予。

### 6.2 区域不匹配
Redshift 与 S3 桶跨区域时，通过 region 参数指定桶所在区域。

### 6.3 时间戳路径覆盖问题
每次运行生成新时间戳子目录，避免使用 ALLOWOVERWRITE 覆盖历史数据。建议在脚本末尾清理过期 S3 临时文件（可用 wr.s3.delete_objects）。

### 6.4 性能优化
- 大数据量时 parallel=True 有效，若需严格顺序则设 parallel=False 配合 ORDER BY。
- 读取超大 df_2 时使用 wr.s3.read_parquet(..., chunked=True) 分块处理。
- 本地 Parquet 读取可加 memory_map=True 降低内存。

## 7. 扩展建议
- 将 Redshift 连接参数、S3 桶名、SQL 等抽取为配置文件或环境变量，提高可移植性。
- 添加 logging 记录 UNLOAD 耗时、数据行数。
- 对 UNLOAD 和 S3 读取加入重试机制应对网络波动。
- 使用 openpyxl 或 xlsxwriter 自定义 Excel 单元格样式（列宽、字体、颜色）。

此方案覆盖了从本地数据、云端仓库、对象存储到最终 Excel 输出的全链路，适用于数据质量比对、多源融合报表等场景。代码已在 Python 3.9+ 及对应库版本下测试通过。

热心网友3

感谢分享，这篇流程写得非常清晰和实用！最近正好在做一个类似的数据导出任务，你的步骤对我帮助很大。特别是用 awswrangler 封装 UNLOAD 并把结果直接读成 DataFrame 那部分，省去了手动处理 S3 文件的麻烦。另外时间戳防覆盖也是个好习惯，我之前踩过坑。 想请教一下：如果 df_2 数据量超大（比如上亿行），用 wr.s3.read_parquet 一次性读入内存会不会撑爆？你提到可以用 chunked=True 分块，那后续写入 Excel 时如何把分块数据分别写入不同 Sheet？还是说这种情况会考虑其他输出格式？谢谢！

热心网友3

这篇教程写得非常清晰实用！从环境搭建到每一步的具体实现都交代得很详细，特别是 UNLOAD 到 S3 再通过 awswrangler 读取的方式，比直接用 COPY 或者一次性拉取全部数据要高效得多，而且自动分片合并读取对内存也很友好。 两个亮点特别值得点赞：一是在 S3 路径里嵌入时间戳避免冲突，二是最后的 `ignore_index=True` 和 `sort_values` 搭配，保证了 Excel 里两个 sheet 的顺序一致。 补充一个小建议：生产环境里可以把敏感信息（密码、IAM Role 等）放到 Secrets Manager 或 AWS Parameter Store 里，而不仅靠环境变量，安全性会更好。另外如果 Redshift 返回的数据量非常大，可以尝试在 `read_parquet` 时加上 `chunked=True` 再逐块合并，避免一次性占用太多内存。 整体流程很完整，适合直接拿过来改改用，感谢分享！

热心网友3

感谢分享这么详细的教程！从本地 Parquet 读取到 Redshift UNLOAD 再到 S3 合并成 DataFrame，最后输出 Excel 多 Sheet，整个流程很完整。尤其喜欢你强调 UNLOAD 的 iam_role 权限和 region 参数，以及用时间戳避免路径覆盖，这些都是实际落地容易踩坑的地方。我有个小问题：如果 df_2 数据量特别大，用 chunked=True 分块读取后，是否还需要额外做一次合并再排序？还是说可以直接对每个分块排序后追加写入？另外，排序前的 id 类型校验（比如统一转为 int64）是否能加个注释，防止因类型不一致报错？整体上已经是非常实用的工程代码了，收藏备用！