文档中心

大数据研发治理套件

Activity （任务节点）类型

Python

Python Activity 用于在 Pipeline 中执行 Workspace 中已有的 Python 文件。您可以先在文件开发页编写和调试 .py 脚本，再将脚本加入 Pipeline，配置正式运行所需的资源、参数和依赖关系。

概述

Python Activity 适合通用数据处理、业务规则计算、文件处理及已有 Python 脚本的自动化编排。一次执行会在所选计算资源组中启动一个 Python 进程。
Python Activity 暂不提供 EMR 队列或 Spark 参数配置。需要 PySpark、交互式多 Cell 开发或 Notebook 能力时，请使用 Notebook。选择 Python 文件不会自动获得 EMR 网络连通性或数据访问权限。

使用前提

已在当前项目的 Workspace 中准备 .py 文件，并具备文件查看和使用权限。
项目中存在可用的计算资源组和运行镜像。
镜像包含脚本需要的 Python 版本；访问数据源时，还需具备相应网络条件和数据权限。
正式运行身份具有脚本、依赖文件及目标数据的相应权限。

配置步骤

在文件开发页打开或创建 .py 文件，编写脚本。
在当前文件的配置区设置计算资源组、镜像、资源和调试参数，运行文件并检查调试记录。
在 Pipeline 的编排页面添加 Python Activity，选择该 .py 文件。
配置 Activity 的正式资源、命令行参数、环境变量和上下游依赖。
检查运行策略，保存并调试 Pipeline；确认结果后发布。
在 Pipeline 实例中选中该 Activity，查看执行状态、输入、输出和日志。

说明

文件调试配置与 Activity 配置分别保存。
首次选择文件时，可用文件调试默认值填充 Activity 的空字段。
Activity 保存后独立维护，后续修改文件调试配置不会自动覆盖已有的 Activity 配置。
代码或正式配置变更后，需要重新发布才能用于新的正式运行。

配置说明

配置项	必填	说明
节点名称	是	在当前 Pipeline 中唯一，建议按业务动作命名，例如 `clean_orders`。命名规则沿用通用 Activity 配置。
描述	否	说明脚本用途。
文件路径	是	选择 Workspace 中的 .py 文件。Activity 中不直接编辑内联代码。
计算资源组	是	承载脚本执行；网络与私有镜像凭证由资源组提供。
镜像地址	否	填写脚本运行所需的镜像地址。Python 版本由镜像决定，当前支持 Python 3.7 和 Python 3.12，具体以页面可选项为准。
资源配置	是	按 CU 选择资源规格，1 CU 对应 1 Core、4 GB 内存。可用规格以所选计算资源组为准。
依赖包	否	填写脚本所需的 Python 包及版本，例如 `包名==版本号`。请确保依赖包与所选镜像中的 Python 版本兼容。
启动参数	否	填写传给脚本的启动参数。该参数可在脚本中读取。
环境变量	否	配置脚本运行所需的环境变量，每行填写一个 `KEY=VALUE`。请勿在代码或配置中保存明文凭证。
产出数据登记	否	声明脚本写入的数据集，供数据血缘与运维检索使用。登记不会自动建表或授予写权限。
运行策略	否	配置失败重试、重试间隔、单次执行超时及超时后是否重试。
依赖资源	否	从 Workspace 选择脚本运行所需的依赖文件。

说明

正式执行前，请核对 Activity 中的文件路径、镜像地址、资源配置、启动参数、环境变量和依赖资源。当前配置页不提供独立的工作目录配置项。

配置示例

以下是 Activity 的基础结构节选。计算资源组、镜像、资源、参数和依赖配置请先通过界面配置，并以界面生成的 YAML 为准；本节选不能直接作为完整发布配置。

activities:
  - name: clean_orders
    type: python
    source: WORKSPACE
    path: /Workspace/Users/demo/scripts/clean_orders.py
    kernelType: python_kubernetes
    position:
      x: "200"
      y: "100"

参数传递

Python Activity 支持通过有序参数向脚本传值。以脚本 clean_orders.py 为例：

import argparse
import json
from datetime import datetime

parser = argparse.ArgumentParser()
parser.add_argument("--biz-date", required=True)
args = parser.parse_args()
datetime.strptime(args.biz_date, "%Y%m%d")

# 在这里添加实际处理逻辑。
print(json.dumps({"biz_date": args.biz_date, "validation": "passed"}))

在 Activity 参数中按顺序配置：

顺序	值
1	`--biz-date`
2	`20261013`

说明

脚本从 argparse 读取参数。参数值可以来自常量、Pipeline 参数、系统变量或有依赖关系的上游 Output，建议通过参数选择器插入引用。
业务日期等参数值应与脚本要求的格式保持一致。
文件开发页中的参数预览仅用于检查调度时间、业务日期与参数转换结果，不会创建调度计划或正式 Pipeline 实例。

输出与运行结果

执行日志记录脚本的标准输出（stdout）和标准错误（stderr）。可通过 print 输出需要检查的日志信息。
在实例详情页选择对应节点后，在其输出页签中，可以查看 output.exit_code、output.stdout 和 output.stderr。
注意
如果脚本打印的是 JSON，系统仍会将其作为标准输出文本展示，不会自动解析为可直接引用的业务字段。
查看运行日志
- 在 Pipeline 运行详情中，可分别查看调度日志与执行日志。
- 每次重跑操作都会产生新的执行尝试，排障时应选择对应的尝试记录。
- 在 Python 文件开发页面的调试记录中，可查看调试概览和调试日志。

常见问题

问题	排查方法
`ModuleNotFoundError`	检查运行镜像、依赖文件及依赖准备日志；确认包与镜像中的 Python 版本兼容。
文件不存在或相对路径报错	检查脚本路径、正式版本及工作目录；确认所需文件存在且运行身份有权限。
本地调试成功，Pipeline 失败	对比文件调试与 Activity 的镜像、资源、参数、依赖及正式运行身份。
无法访问 EMR 数据	分别检查计算资源组的网络和数据访问权限；PySpark 场景使用 Notebook。
打印 JSON 后下游取不到字段	检查该版本的 Output 采集说明；打印日志不等于完成结构化 Output 配置。
执行超时或资源不足	检查日志和资源使用情况，调整 CPU、内存或超时；同时检查重试是否会重复写入数据。

使用建议

先用文件调试验证脚本，再用 Pipeline 调试验证依赖关系和正式配置。
依赖和解释器版本通过镜像及依赖文件统一管理；不要依赖开发环境临时安装的包。
对写入、上传等操作设计幂等处理，避免重试产生重复数据。
不要在日志中打印密码、Token 或 AK/SK。

大数据研发治理套件

概述 #

使用前提 #

配置步骤 #

配置说明 #

配置示例 #

参数传递 #

输出与运行结果 #

常见问题 #

使用建议 #

相关文档 #

概述