如何在Kedro Node中使用SQL Server Bulk Insert加载S3的CSV数据

1. 获取catalog中bulk_insert_input对应的S3路径 #

Kedro支持在节点中传入数据集实例而非加载后的数据，无需修改现有catalog配置，仅调整节点参数的类型标注即可直接拿到S3路径：

• 在节点函数中给对应参数加kedro.io.Dataset类型标注，Kedro会直接传入CSV数据集实例而非自动加载为DataFrame
• 通过数据集实例的_filepath属性即可拿到完整的S3路径

注意：如果你的SQL Server未配置S3访问权限，可以给bulk_insert_input增加本地缓存配置，Kedro会自动将S3文件同步到本地路径供BULK INSERT使用：

bulk_insert_input:
   type: pandas.CSVDataSet
   filepath: s3://your_bucket/data/02_intermediate/company/motorbikes.csv
   credentials: dev_s3
   cache_dir: data/01_tmp # 自定义本地临时缓存目录

同步后的本地路径同样可以通过_filepath属性获取。

2. 合规获取dw_dev_credentials凭证 #

推荐直接复用catalog中已配置的flp_test数据集能力，不需要手动读取凭证、创建连接，完全符合Kedro规范：

• 把flp_test也作为节点的输入参数传入，同样加kedro.io.Dataset类型标注
• 直接调用该数据集内置的SQLAlchemy引擎获取连接，无需暴露凭证明文

示例完整代码 #

节点逻辑（nodes.py）

from kedro.io import Dataset

def bulk_insert_to_sql(csv_dataset: Dataset, sql_dataset: Dataset):
    # 1. 拿到CSV文件路径（S3路径或本地缓存路径）
    csv_file_path = str(csv_dataset._filepath)
    # 2. 直接从SQL数据集配置中拿到完整表名，避免硬编码
    schema = sql_dataset.save_args.get("schema", "dbo")
    full_table_name = f"[{schema}].[{sql_dataset.table_name}]"

    # 3. 复用现有连接执行BULK INSERT
    with sql_dataset._engine.connect() as conn:
        # 用参数化查询避免SQL注入风险
        insert_sql = f"BULK INSERT {full_table_name} FROM ? WITH (FORMAT = 'CSV')"
        conn.execute(insert_sql, parameters=(csv_file_path,))
        conn.commit()
    return

流水线注册（pipeline.py）

from kedro.pipeline import Pipeline, node
from .nodes import bulk_insert_to_sql

def create_pipeline(**kwargs):
    return Pipeline(
        [
            node(
                func=bulk_insert_to_sql,
                # 直接传入catalog中定义的两个数据集即可
                inputs=["bulk_insert_input", "flp_test"],
                outputs=None,
                name="bulk_insert_s3_to_sql_server"
            )
        ]
    )

优化提示：如果你的CSV包含表头、自定义分隔符等特殊配置，可以直接在BULK INSERT的WITH参数中补充对应的配置项，也可以把这些配置放到数据集的save_args/load_args中统一管理。

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