如何获取Dataprep作业的Dataflow模板以移除Dataprep依赖

优先方案：从历史运行记录提取作业配置（零逻辑重写） #

Dataprep所有作业本质都是调用GCP原生Dataflow Flex模板运行，哪怕官方下线了模板导出功能，你也可以直接从历史运行的Dataflow作业里拿到完整可复用的启动配置，和原来导出的模板内容完全一致：

• 打开GCP控制台进入Dataflow作业列表，筛选标签关联了Dataprep/Trifacta的历史作业，找最近成功运行、处理逻辑和当前生产一致的作业条目
• 进入作业详情页，复制全量启动配置：包括Flex模板的GCS路径、所有传入参数（输入输出路径、清洗规则配置、机器规格、网络/权限配置、异常处理规则）
• 如果找不到可用的历史作业，手动触发一次现有Dataprep作业，等Dataflow侧生成作业记录后，立刻去详情页抓配置即可，不需要等作业跑完
• 手动用复制到的配置触发一次Dataflow作业，对比输出结果和原Dataprep作业的输出，确认字段、数据量、清洗规则完全匹配再往下走

替换原有触发链路 #

拿到可用的Dataflow启动配置后，直接把Dataprep从链路里剔除：

• 删除原有通过Python调用Dataprep API触发作业的逻辑，替换为直接调用Dataflow Flex模板启动接口的逻辑，Python版最简触发代码参考：

from google.cloud import dataflow_v1beta3

def gcs_trigger(event, context):
    # 监听GCS文件上传事件触发
    uploaded_file = event
    input_gcs_path = f"gs://{uploaded_file['bucket']}/{uploaded_file['name']}"
    client = dataflow_v1beta3.FlexTemplatesServiceClient()

    launch_req = dataflow_v1beta3.LaunchFlexTemplateRequest(
        project_id="你的GCP项目ID",
        location="作业部署区域，比如us-central1",
        launch_parameter=dataflow_v1beta3.LaunchFlexTemplateParameter(
            job_name=f"migrated-dataprep-job-{uploaded_file['timeCreated'].replace(':','-')}",
            container_spec_gcs_path="从历史作业复制的Flex模板GCS路径",
            parameters={
                "inputFilePattern": input_gcs_path,
                # 其余参数全部从历史作业配置中复制补全
                # 包括输出路径、清洗规则配置、执行参数等
            }
        )
    )
    client.launch_flex_template(request=launch_req)

• 触发器层直接保留原有GCS存储桶的事件通知规则即可，把触发目标从原来的Dataprep调用逻辑换成上面的Dataflow触发逻辑，支持用Cloud Functions、Cloud Run或者你原有部署的Python服务承载触发逻辑
• 权限调整：给触发逻辑用的服务账号分配Dataflow作业启动权限、源存储桶读权限、目标存储/数仓写权限即可，移除所有Dataprep相关的角色授权

迁移注意事项 #

• 不要使用第三方逆向导出的模板：这类模板会丢失Dataprep内置的隐式清洗规则（比如类型自动转换、脏数据过滤规则），跑出来的结果和原逻辑大概率有偏差
• 建议双跑验证1-2周：同一批输入文件同时触发原Dataprep作业和新的Dataflow作业，对比输出结果完全一致后，再正式下线Dataprep服务
• 原Dataprep版本的模板导出窗口参考如下，你从历史作业提取的配置和当年导出功能拿到的模板参数无差异：
  

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