下面结合Python及相关库pandas、redshift_connector、awswrangler、openpyxl、boto3、pathlib和datetime实现从本地读取Parquet文件、从Amazon Redshift导出数据到S3并下载、合并两个DataFrame并按指定列排序后导出至Excel文件一个文件多个Sheet的完整数据工程流程并展开详细论述。所有涉及的库与实现思路均可在生产环境或数据分析项目中直接复用。1. 环境准备首先安装所需依赖包并导入相关库pip install pandas pyarrow redshift_connector awswrangler openpyxl boto3importosimportpandasaspdimportredshift_connectorimportawswrangleraswrimportboto3fromdatetimeimportdatetimefrompathlibimportPath2. 从本地Parquet读取DataFramedf_1利用pandas.read_parquet从本地读取parquet文件并将其加载为DataFrameparquet_path./data/source_data.parquetdf_1pd.read_parquet(parquet_path,enginepyarrow)read_parquet默认使用pyarrow引擎该引擎对列式存储的parquet文件有较好的性能表现。在处理大规模数据时可通过设置use_threadsTrue启用多线程读取或通过memory_mapTrue使用内存映射文件以提升读取速度。3. 从Amazon Redshift通过UNLOAD导出数据到S3并下载为DataFramedf_2此部分整合了redshift_connector、awswrangler、boto3、pathlib和datetime等库的功能。3.1 建立Redshift连接使用redshift_connector创建与Amazon Redshift集群的连接connredshift_connector.connect(hostos.getenv(REDSHIFT_HOST),databaseos.getenv(REDSHIFT_DATABASE),portint(os.getenv(REDSHIFT_PORT,5439)),useros.getenv(REDSHIFT_USER),passwordos.getenv(REDSHIFT_PASSWORD))此方式基于AWS凭证进行认证符合Amazon Redshift Python连接器的标准用法。3.2 利用awswrangler执行UNLOAD操作awswrangler.redshift.unload_to_files封装了Redshift的UNLOAD命令可将查询结果直接导出至S3支持CSV或Parquet格式适用于大规模数据导出场景。使用带时间戳的S3路径以规避文件名冲突timestampdatetime.now().strftime(%Y%m%d_%H%M%S)s3_unload_pathfs3://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(sqlsql_query,paths3_unload_path,conconn,iam_roleos.getenv(REDSHIFT_IAM_ROLE),unload_formatPARQUET,parallelTrue,regionos.getenv(AWS_REGION,us-east-1))UNLOAD路径设计说明通过在S3路径中嵌入datetime.now().strftime(%Y%m%d_%H%M%S)生成的时间戳可确保每次运行生成唯一路径防止多次执行之间的文件覆盖。也可在UNLOAD命令中使用ALLOWOVERWRITE参数覆盖已有文件但通过时间戳隔离版本是更为稳健的做法。参数说明sql执行的查询语句支持任意复杂的SELECT语句。pathS3目标路径UNLOAD会在此路径下生成一个或多个数据文件。con通过redshift_connector建立的连接对象。iam_role具有写入目标S3桶权限的IAM角色ARN该角色至少需要s3:PutObject和s3:ListBucket权限。unload_format导出格式支持CSV或PARQUET此处选用PARQUET以获得更高的压缩率和更快的后续读取速度。parallel是否并行写入多个文件默认True可充分利用Redshift集群的切片数提升导出速度。regionS3桶所在的AWS区域。3.3 从S3下载并读取为DataFrameUNLOAD执行完毕后使用boto3列出S3路径下生成的文件并通过awswrangler.s3.read_parquet直接读取S3上的parquet文件为DataFramesessionboto3.Session(region_nameos.getenv(AWS_REGION,us-east-1))df_2wr.s3.read_parquet(paths3_unload_path,boto3_sessionsession)awswrangler.s3.read_parquet可以自动识别S3路径下的所有parquet文件并将它们合并为单个DataFrame极大简化了S3数据读取流程。3.4 关闭Redshift连接conn.close()4. 统一按id列排序两个DataFrame均按照id列升序排列以确保写入Excel后的数据具备一致性df_1df_1.sort_values(byid,ascendingTrue,ignore_indexTrue)df_2df_2.sort_values(byid,ascendingTrue,ignore_indexTrue)ignore_indexTrue可重置索引避免将旧索引带入Excel输出。5. 将两个DataFrame写入同一Excel文件的不同Sheet使用pd.ExcelWriter配合openpyxl引擎支持.xlsx格式将两个DataFrame分别写入不同Sheetoutput_pathPath(./output/mult_sheets_1.xlsx)output_path.parent.mkdir(parentsTrue,exist_okTrue)withpd.ExcelWriter(output_path,engineopenpyxl)aswriter:df_1.to_excel(writer,sheet_namedf_1,indexFalse)df_2.to_excel(writer,sheet_namedf_2,indexFalse)with语句可确保Excel文件在写入完成后自动保存并关闭。sheet_name分别设置为df_1和df_2indexFalse避免将行索引写入Excel。如需在同一文件内写入更多DataFrame只需继续调用to_excel方法即可。6. 完整代码示例将上述各模块整合为完整可运行的代码importosimportpandasaspdimportredshift_connectorimportawswrangleraswrimportboto3fromdatetimeimportdatetimefrompathlibimportPath# ---------- 配置 ----------parquet_path./data/source_data.parquetoutput_pathPath(./output/mult_sheets_1.xlsx)output_path.parent.mkdir(parentsTrue,exist_okTrue)sql_query SELECT id, column_a, column_b, created_at FROM your_schema.your_table WHERE created_at 2026-01-01 # ---------- 1. 读取本地Parquetdf_1 ----------df_1pd.read_parquet(parquet_path,enginepyarrow)# ---------- 2. 连接Redshift ----------connredshift_connector.connect(hostos.getenv(REDSHIFT_HOST),databaseos.getenv(REDSHIFT_DATABASE),portint(os.getenv(REDSHIFT_PORT,5439)),useros.getenv(REDSHIFT_USER),passwordos.getenv(REDSHIFT_PASSWORD))# ---------- 3. UNLOAD到S3 ----------timestampdatetime.now().strftime(%Y%m%d_%H%M%S)s3_unload_pathfs3://your-bucket/redshift_unload/{timestamp}/wr.redshift.unload_to_files(sqlsql_query,paths3_unload_path,conconn,iam_roleos.getenv(REDSHIFT_IAM_ROLE),unload_formatPARQUET,parallelTrue,regionos.getenv(AWS_REGION,us-east-1))# ---------- 4. 从S3读取为DataFramedf_2 ----------sessionboto3.Session(region_nameos.getenv(AWS_REGION,us-east-1))df_2wr.s3.read_parquet(paths3_unload_path,boto3_sessionsession)# ---------- 5. 关闭Redshift连接 ----------conn.close()# ---------- 6. 按id排序 ----------df_1df_1.sort_values(byid,ascendingTrue,ignore_indexTrue)df_2df_2.sort_values(byid,ascendingTrue,ignore_indexTrue)# ---------- 7. 写入Excel ----------withpd.ExcelWriter(output_path,engineopenpyxl)aswriter:df_1.to_excel(writer,sheet_namedf_1,indexFalse)df_2.to_excel(writer,sheet_namedf_2,indexFalse)print(fExcel文件已生成{output_path})7. 常见问题与调试建议7.1 UNLOAD权限错误执行UNLOAD时若遇到AccessDenied错误请检查iam_role对应的IAM角色是否具备以下权限s3:PutObject允许写入S3对象s3:ListBucket允许列出桶内对象7.2 区域不匹配若S3桶与Redshift集群位于不同AWS区域需通过region参数明确指定S3桶所在区域否则UNLOAD将失败。7.3 时间戳路径覆盖问题如需在同一路径下多次运行建议保留时间戳前缀避免使用ALLOWOVERWRITE参数以防止意外覆盖历史数据。同时可在脚本末尾添加S3清理逻辑定期删除过期的UNLOAD临时文件以控制存储成本。7.4 性能优化建议UNLOAD并行度parallelTrue可显著提升大数据量的导出速度但如果后续处理对文件顺序有严格要求可设置parallelFalse并按ORDER BY子句生成单一有序文件。Parquet读取若df_2数据量极大可使用wr.s3.read_parquet的chunkedTrue参数实现分块读取降低内存压力。内存映射对于本地parquet文件的读取可通过pd.read_parquet(..., memory_mapTrue)减少内存开销。8. 扩展建议以上流程可作为日常数据分析或ETL任务的基础模板。如需进一步工程化可考虑参数化配置将S3桶名、Redshift连接信息、SQL查询等提取为配置文件或环境变量提升代码的可移植性。日志记录引入Python内置的logging模块记录UNLOAD耗时、DataFrame行数等关键指标。错误重试机制对UNLOAD和S3读取操作添加重试逻辑增强脚本在临时网络波动下的鲁棒性。S3临时文件清理UNLOAD完成后可选择性地删除S3上的临时文件以控制存储成本可通过wr.s3.delete_objects实现。多Sheet样式定制若需要对Excel输出进行样式设置如列宽、字体、颜色可使用openpyxl或xlsxwriter的高级接口对每个Sheet单独定制。此方案覆盖了从本地数据读取、云端数据仓库导出、云存储中转、本地合并到最终Excel输出的全链路适用于数据分析报告生成、数据质量比对、多源数据融合等业务场景。