Kedro中行级API调用丰富CSVDataSet的实现方案咨询

你的思路确实存在问题，APIDataSet并不适合这种逐行动态发起API请求的场景，更推荐在Node内部直接处理API调用。

为什么APIDataSet不可行？ #

APIDataSet的核心设计是「静态配置、一次性加载」：

• 它的请求参数（URL、请求头、查询参数等）在catalog中预先固定，运行时无法动态修改
• load()方法执行的是预定义的单一请求，无法针对CSV的每一行生成不同的API调用，自然没法把每行的参数传递进去

正确的实现方式 #

直接在Node内部完成API调用逻辑，步骤如下：

1. 从CSVDataSet加载完整数据集
2. 遍历数据的每一行，提取用于API请求的关键参数（比如用户ID、商品SKU等）
3. 针对每个参数构造对应的API请求（可以用requests这类HTTP库），获取返回数据
4. 将API返回的字段与原行数据合并，生成丰富后的行记录
5. 把所有处理后的记录统一保存到SQLTableDataSet

示例伪代码 #

def run(self, context):
    # 加载CSV数据
    csv_data = context.catalog.load("your_csv_dataset")
    
    enriched_rows = []
    for index, row in csv_data.iterrows():
        # 提取行内参数
        api_param = row["target_column"]
        # 发起API请求
        api_response = requests.get(f"https://your-api-url/{api_param}")
        api_response.raise_for_status()
        api_data = api_response.json()
        
        # 合并数据
        enriched_row = {**row.to_dict(), **api_data}
        enriched_rows.append(enriched_row)
    
    # 转换为DataFrame并保存到SQL
    enriched_df = pd.DataFrame(enriched_rows)
    context.catalog.save("your_sql_dataset", enriched_df)

额外优化建议 #

• 如果API支持批量查询，尽量批量发起请求，减少调用次数，提升效率
• 加入异常捕获逻辑，处理API请求失败的情况（比如超时、返回错误码），避免整个流程中断
• 对失败的请求加入重试机制，应对临时的服务不可用问题

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