Airflow中将GCS的CSV数据写入PostgreSQL的最Pythonic方式

结合Python生态风格、Airflow最佳实践和生产可用性，三种方案的优先级从高到低如下：

首选：Airflow原生算子组合（方案3落地实现） #

这是最符合Python「不要重复造轮子」「声明式优于命令式」设计原则的方案，所有连接管理、重试、报错兜底逻辑都由Airflow官方维护的算子封装，无需手写冗余的底层IO代码，可维护性极高。
打通GCS和PostgreSQL的逻辑非常简单，分两步即可实现：

• 先用GCSToLocalFilesystemOperator将GCS上的CSV文件下载到Airflow worker的本地临时目录
• 再用PostgresOperator执行PostgreSQL原生的COPY FROM命令，直接读取本地CSV批量插入表，性能比逐行插入高2个数量级以上

参考代码片段：

from airflow.providers.google.cloud.transfers.gcs_to_local import GCSToLocalFilesystemOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime

with DAG('gcs_csv_to_cloudsql', start_date=datetime(2024,1,1), schedule=None) as dag:
    download_csv = GCSToLocalFilesystemOperator(
        task_id="download_gcs_csv",
        bucket="你的GCS桶名",
        object_name="csv文件在桶内的路径",
        filename="/tmp/temp_data.csv",
        gcp_conn_id="你配置的GCP连接ID",
    )

    import_to_pg = PostgresOperator(
        task_id="import_csv_to_postgres",
        postgres_conn_id="你配置的Cloud SQL连接ID",
        sql="""
        COPY 目标表名(列1, 列2, 列3)
        FROM '/tmp/temp_data.csv'
        DELIMITER ','
        CSV HEADER;
        """,
    )

    download_csv >> import_to_pg

只要Cloud SQL和Airflow集群网络连通（同VPC或者开公网白名单），这个方案可以直接跑通，没有额外依赖。

次选：Pandas导入方案 #

如果CSV需要先做数据清洗、类型转换再入库，选Pandas是最合适的，代码简洁可读性高，是Python数据场景的标准写法。
实现逻辑：

• 借助gcsfs库可以直接读取GCS上的CSV到DataFrame，无需手动下载到本地
• 调用df.to_sql()方法直接写入PostgreSQL，底层自动封装了SQLAlchemy批量插入逻辑，不用自己手写插入语句

参考代码片段：

import pandas as pd
from sqlalchemy import create_engine
from airflow.decorators import task

@task
def import_csv_with_pandas():
    # 直接读取GCS上的CSV
    df = pd.read_csv("gs://桶名/文件路径.csv")
    # 此处可添加任意数据清洗、字段转换逻辑
    pg_engine = create_engine("postgresql+psycopg2://用户名:密码@实例地址:端口/库名")
    df.to_sql(
        name="目标表名",
        con=pg_engine,
        if_exists="append",
        index=False,
        chunksize=1000
    )

这个方案的优势是灵活度高，需要做数据转换时比原生算子便捷很多。

不推荐：纯SQLAlchemy逐行插入 #

除非有特殊的逐行校验、异常处理需求，否则完全没必要自己手写遍历CSV、拼接插入语句的逻辑，代码冗余度高、易出BUG、性能差，不符合Python尽量复用成熟工具的风格。

内容的提问来源于stack exchange，提问作者David Masip