Python Redshift UNLOAD 导出 Parquet 并写
在数据工程与数据分析场景中，经常需要将本地 Parquet 文件与 Amazon Redshift 中的查询结果合并，并输出为单个 Excel 文件的不同 Sheet。本文基于 Python 生态中的 pandas、redshift_connector、awswrangler 和 openpyxl，实现一套完整的从本地读取、Redshift UNLOAD 至 S3、下载合并、排序再到 Excel 多 Sheet 写入的自动化流程。## 环境与依赖
首先安装必要的 Python 包：
pip install pandas pyarrow redshift_connector awswrangler openpyxl boto3
导入时依次引入：
import os
import pandas as pd
import redshift_connector
import awswrangler as wr
import boto3
from datetime import datetime
from pathlib import Path
## 1. 从本地 Parquet 读取 DataFrame（df_1）
使用 pandas 的 read_parquet 方法，引擎指定为 pyarrow（默认），对大列式存储文件性能更优。可通过 use_threads=True 启用多线程读取，memory_map=True 使用内存映射降低 I/O 开销。
parquet_path = "./data/source_data.parquet"
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 方法，从环境变量中读取主机、数据库、端口、用户名和密码：
conn = redshift_connector.connect(
host=os.getenv("REDSHIFT_HOST"),
database=os.getenv("REDSHIFT_DATABASE"),
port=int(os.getenv("REDSHIFT_PORT", 5439)),
user=os.getenv("REDSHIFT_USER"),
password=os.getenv("REDSHIFT_PASSWORD")
)
### 2.2 执行 UNLOAD 到 S3
利用 awswrangler 的 wr.redshift.unload_to_files 封装 Redshift 的 UNLOAD 命令。为避免路径覆盖，在 S3 路径中嵌入时间戳：
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
s3_unload_path = f"s3://your-bucket/redshift_unload/{timestamp}/"
sql_query = """
SELECT id, column_a, column_b, created_at
FROM your_schema.your_table
WHERE created_at >= '2026-01-01'
"""
wr.redshift.unload_to_files(
sql=sql_query,
path=s3_unload_path,
con=conn,
iam_role=os.getenv("REDSHIFT_IAM_ROLE"),
unload_format="PARQUET",
parallel=True,
region=os.getenv("AWS_REGION", "us-east-1")
)
关键参数说明：
- 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 文件并合并读取：
session = boto3.Session(region_name=os.getenv("AWS_REGION", "us-east-1"))
df_2 = wr.s3.read_parquet(path=s3_unload_path, boto3_session=session)
如果 df_2 数据量极大，可使用 chunked=True 分块读取以控制内存。
### 2.4 关闭连接
conn.close()
## 3. 统一按 id 列排序
为保证 Excel 中两个 Sheet 的数据顺序一致，对两个 DataFrame 均按 id 升序排列，并重置索引：
df_1 = df_1.sort_values(by="id", ascending=True, ignore_index=True)
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 语句确保文件自动保存关闭：
output_path = Path("./output/mult_sheets_1.xlsx")
output_path.parent.mkdir(parents=True, exist_ok=True)
with pd.ExcelWriter(output_path, engine="openpyxl") as writer:
df_1.to_excel(writer, sheet_name="df_1", index=False)
df_2.to_excel(writer, sheet_name="df_2", index=False)
index=False 防止行索引写入。若需更多 Sheet 继续调用 to_excel 即可。
## 5. 完整代码整合
将上述片段合并为可独立运行的脚本：
import os
import pandas as pd
import redshift_connector
import awswrangler as wr
import boto3
from datetime import datetime
from pathlib import Path
# 配置
parquet_path = "./data/source_data.parquet"
output_path = Path("./output/mult_sheets_1.xlsx")
output_path.parent.mkdir(parents=True, exist_ok=True)
sql_query = """
SELECT id, column_a, column_b, created_at
FROM your_schema.your_table
WHERE created_at >= '2026-01-01'
"""
# 1. 读取本地 Parquet
df_1 = pd.read_parquet(parquet_path, engine="pyarrow")
# 2. 连接 Redshift
conn = redshift_connector.connect(
host=os.getenv("REDSHIFT_HOST"),
database=os.getenv("REDSHIFT_DATABASE"),
port=int(os.getenv("REDSHIFT_PORT", 5439)),
user=os.getenv("REDSHIFT_USER"),
password=os.getenv("REDSHIFT_PASSWORD")
)
# 3. UNLOAD 到 S3
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
s3_unload_path = f"s3://your-bucket/redshift_unload/{timestamp}/"
wr.redshift.unload_to_files(
sql=sql_query,
path=s3_unload_path,
con=conn,
iam_role=os.getenv("REDSHIFT_IAM_ROLE"),
unload_format="PARQUET",
parallel=True,
region=os.getenv("AWS_REGION", "us-east-1")
)
# 4. 从 S3 读取
df_2 = wr.s3.read_parquet(path=s3_unload_path, boto3_session=boto3.Session(region_name=os.getenv("AWS_REGION", "us-east-1")))
# 5. 关闭连接
conn.close()
# 6. 排序
df_1 = df_1.sort_values(by="id", ascending=True, ignore_index=True)
df_2 = df_2.sort_values(by="id", ascending=True, ignore_index=True)
# 7. 写入 Excel
with pd.ExcelWriter(output_path, engine="openpyxl") as writer:
df_1.to_excel(writer, sheet_name="df_1", index=False)
df_2.to_excel(writer, sheet_name="df_2", index=False)
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+ 及对应库版本下测试通过。
Re: Python Redshift UNLOAD 导出 Parquet 并写
感谢分享，这篇流程写得非常清晰和实用！最近正好在做一个类似的数据导出任务，你的步骤对我帮助很大。特别是用 awswrangler 封装 UNLOAD 并把结果直接读成 DataFrame 那部分，省去了手动处理 S3 文件的麻烦。另外时间戳防覆盖也是个好习惯，我之前踩过坑。 想请教一下：如果 df_2 数据量超大（比如上亿行），用 wr.s3.read_parquet 一次性读入内存会不会撑爆？你提到可以用 chunked=True 分块，那后续写入 Excel 时如何把分块数据分别写入不同 Sheet？还是说这种情况会考虑其他输出格式？谢谢！Re: Python Redshift UNLOAD 导出 Parquet 并写
这篇教程写得非常清晰实用！从环境搭建到每一步的具体实现都交代得很详细，特别是 UNLOAD 到 S3 再通过 awswrangler 读取的方式，比直接用 COPY 或者一次性拉取全部数据要高效得多，而且自动分片合并读取对内存也很友好。 两个亮点特别值得点赞：一是在 S3 路径里嵌入时间戳避免冲突，二是最后的 `ignore_index=True` 和 `sort_values` 搭配，保证了 Excel 里两个 sheet 的顺序一致。 补充一个小建议：生产环境里可以把敏感信息（密码、IAM Role 等）放到 Secrets Manager 或 AWS Parameter Store 里，而不仅靠环境变量，安全性会更好。另外如果 Redshift 返回的数据量非常大，可以尝试在 `read_parquet` 时加上 `chunked=True` 再逐块合并，避免一次性占用太多内存。 整体流程很完整，适合直接拿过来改改用，感谢分享！Re: Python Redshift UNLOAD 导出 Parquet 并写
感谢分享这么详细的教程！从本地 Parquet 读取到 Redshift UNLOAD 再到 S3 合并成 DataFrame，最后输出 Excel 多 Sheet，整个流程很完整。尤其喜欢你强调 UNLOAD 的 iam_role 权限和 region 参数，以及用时间戳避免路径覆盖，这些都是实际落地容易踩坑的地方。我有个小问题：如果 df_2 数据量特别大，用 chunked=True 分块读取后，是否还需要额外做一次合并再排序？还是说可以直接对每个分块排序后追加写入？另外，排序前的 id 类型校验（比如统一转为 int64）是否能加个注释，防止因类型不一致报错？整体上已经是非常实用的工程代码了，收藏备用！
页:
[1]