可以用transform装饰器实现，但不能通过循环调用Input函数的方式动态绑定数据集——因为Input是在装饰器初始化阶段静态声明的依赖，运行时的循环调用无法触发新的数据集绑定。

你之前分模块方案的问题原因 #

读数据集模块用Input接收ID读取的方法不生效，核心原因是：Input的依赖关系是在模块加载时就固定的，不是运行时根据传入的ID动态生成的。即便你传入不同ID，该模块的Input依然绑定初始化时的数据集，无法动态切换。

可行解决方法 #

直接在主模块中读取存储ID的数据集后，调用平台提供的数据集读取API（而非Input）来动态获取每个数据集的DataFrame，示例代码如下：

import pandas as pd
from your_platform.sdk import dataset_client  # 替换为你使用的平台SDK
from transform import transform, Input, Output

@transform(
    inputs=Input("存储数据集ID的主数据集"),
    outputs=Output("处理后的合并数据集")
)
def dynamic_read_datasets(inputs, outputs):
    # 1. 读取存储ID的数据集
    id_dataset = inputs["存储数据集ID的主数据集"].to_df()
    # 提取前100个数据集ID（根据业务调整）
    target_ids = id_dataset["dataset_id_column"].tolist()[:100]
    
    # 2. 循环读取每个数据集
    all_dataframes = []
    for ds_id in target_ids:
        # 用平台API直接读取指定ID的数据集并转为DataFrame
        target_dataset = dataset_client.get_dataset(ds_id)
        df = target_dataset.to_pandas()  # 替换为平台对应的转DataFrame方法
        all_dataframes.append(df)
    
    # 3. 按业务需求处理所有数据集（示例为合并）
    merged_result = pd.concat(all_dataframes, ignore_index=True)
    # 写入输出数据集
    outputs["处理后的合并数据集"].write(merged_result)

补充说明 #

如果你的场景必须依赖transform的Input机制做依赖追踪，只能提前获取所有数据集ID，动态生成装饰器的inputs参数，但这种方式不适合从数据集动态读取ID的场景——因为装饰器初始化时还未读取到ID列表，所以更推荐上述直接调用API的方案。

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