di-agent-flow-pyflow

Pyflow API 规范

路由由其他部分决定。 请求应通过 pyflow 还是 DataStage SDK 编写，以及这是创建还是编辑，由 di-agent-flow-lifecycle 技能（AUTHOR 状态）决定。本文档是 pyflow 语言参考及其使用方法。开始处理流之前，必须加载 di-agent-flow-lifecycle 技能。在加载生命周期技能之前，不要编写任何 Pyflow 代码。

使用指南

Pyflow 表达的是声明式意图；pyflow 编译器会将 DSL 降低为引擎专用的命令式流，为你声明的内容生成准确的功能计划。你表达用户的目标；编译器负责选择各个阶段并确定它们之间的连接方式。因此，使用 pyflow 无需掌握 DataStage 各阶段的专业知识。

编译器会验证你的流，并在发布任何资产之前提供详细的编译时反馈，确保正确性，还会为你设置连接元数据——所需令牌数远少于 SDK。最终生成的是原生 pyflow 流，后续编辑时也能顺利往返转换。

Pyflow 的价值在于构建结构——源、联接、筛选器及其间的连接方式，这是在 SDK 中手动编写时最费力且最容易出错的部分。一旦结构建好，在 SDK 中添加表达式和阶段属性就很简单了。

不要因为某个函数或阶段不受支持就放弃 pyflow。 假设某个流需要 pyflow 不具备的正则表达式提取功能，仍应从 pyflow 开始：先构建源、筛选器和联接，再将缺失部分插入生成的 SDK 代码中。从头手动编写新的联接或源是 SDK 最不可靠的做法——只要 pyflow 能生成框架，就绝不要这样做。

创建流与覆盖流

create_pyflow 有两种模式，由 replace_flow_id 选择：

• 省略 replace_flow_id — 创建一个新流。仅在目标流不存在时使用。
• replace_flow_id=<flow_id> — 重新编译 DSL 并就地覆盖该流，保留其 ID 和名称。对已有流进行重做、修复或重新生成时，就使用此方式。由于 ID 会保留，指向该流的作业会继续运行，也不会创建重复资产。

若要“重做”已有流，不要在未指定 replace_flow_id 的情况下再次调用 create_pyflow，否则会生成第二个资产。绝不要为了重新创建流而删除它。

先搭建框架，再插入细节

当 pyflow 能构建主干结构，但无法处理最后的细节时：

1. 使用 create_pyflow 在 pyflow 中搭建结构——源、筛选器、联接及连接方式。
2. 使用 retrieve_datastage_flow_code 读取生成的 SDK 代码。
3. 精确编辑该 SDK 代码，添加 pyflow 无法表达的内容（表达式、属性、额外的自定义阶段 / Buildop / Java 阶段），然后通过 update_datastage_flow 重新提交完整代码。

SDK 代码冗长——涉及手动连接节点、传播架构，以及可见和隐藏属性。没有可用参考就从头编写结构，极易出错；pyflow 会正确生成连接好的结构，只需进行少量局部修改。

pyflow 的适用边界

已知缺口集中在一处：di-agent-flow-lifecycle/references/registry.md 中本后端的“已知缺口”标题下，分为阻塞性缺口（pyflow 无法搭建可用主干结构——创建时改用 SDK）和可插入式缺口（pyflow 能构建流，之后再添加一个局部阶段或属性）。这是路由决策，应由路由器负责，而非本文档。

本文档有意不列出 pyflow 支持的功能。pyflow 是声明式的——你表达目标，由编译器选择阶段——因此，支持阶段列表无法保持准确，因引擎而异，还容易引导人们提出错误的问题。真正要问的是 pyflow 能否表达请求；下方的引擎操作表会直接回答这个问题。

编辑现有流

• 添加或更改结构（新来源、联接或不同形状）-> 在 pyflow 中搭建该结构，然后对生成的 SDK 进行精细编辑，补上 pyflow 无法表达的内容——表达式、属性，甚至额外的自定义 / Buildop / Java 阶段。pyflow 能可靠地搭建主干；之后再把其余部分拼接进去。
• 仅修改现有结构中的表达式或属性，不添加新连线 -> 直接编辑检索到的 SDK。
• 核心形状没有忠实对应的 pyflow 形式，因此搭建出的框架需要你重新连线，而不是在其上追加内容 -> 先搭建 pyflow 能生成的最接近形状，再通过 update_datastage_flow 重塑生成的 SDK。单独缺少某个阶段不算“没有 pyflow 形式”，因为阶段可以在之后（上方）插入。

Streamsets / Jetstream 引擎说明：di-agent-flow-pyflow-streamsets

pyflow 的主要引擎是 DataStage。本文档介绍如何为此目的使用 pyflow。如果你想编写使用 Streamsets（Jetstream）引擎的流，还需阅读辅助技能 di-agent-flow-pyflow-streamsets，其中介绍了为该引擎编写流时，Streamsets 特有的功能以及 pyflow 语言的差异。如果没有该辅助技能，Streamsets/Jetstream 引擎可能未启用。

代码结构

运行时会提供 q；不要导入或实例化它。每个流都必须：

1. 使用 q.source() 声明来源——只列出被引用的列，并使用资产元数据中的精确名称和类型。必须至少提供一列——q.source("sym") 不带列会导致编译错误。
2. 恰好调用一次 q.name("<snake_case_name>")。
3. 最后恰好使用一个接收器：q.output(frame)；或者在写入目标资产时使用 q.write(frame, "symbol", operation="insert" | "overwrite" | "update" | "create")。operation="create" 仅适用于 DataStage。

代码中不得包含导入语句或 print()。

q 变量——不要重新赋值

q 是全局 DSL 句柄。绝不要将来源结果赋回 q。

# ✗ WRONG — q is now a Frame; q.col(), q.name(), q.output() all break
q = q.source("orders", id="i64", amount="f64")

# ✓ CORRECT — source result goes to a new variable
orders = q.source("orders", id="i64", amount="f64")
q.name("my_flow")
q.output(orders, name="out")

q 仅具有 q 命名空间部分列出的方法。它没有 .lit()、.sum()、.avg()、.max()、.round()、.row_number()、.select() 或其中未列出的任何其他方法。聚合函数、窗口函数和标量运算都属于 Expr 对象；这些对象由 q.col(...) 返回，而不是由 q 本身提供。

引擎目标

调用方会将目标引擎传递给 create_pyflow(engine=...)；不要在代码中声明引擎。引擎决定允许使用哪些 Frame 操作。

DataStage 允许的操作：

• q.source()（每个流可有多个）
• q.output() / q.sink() / q.write()
• .filter()、.sort()
• .select() / .with_columns()
• .head() / .fetch()、.unique()
• .union()、.intersect()
• .group_by().agg()
• .join()、.cross()

符号与绑定

传递给 q.source()、.lookup() 和 q.write() 的字符串是本地符号。调用方通过 create_pyflow(bindings=...) 将每个符号绑定到数据源；符号无需与目录名称一致。每个使用的符号都必须绑定。

将每个符号命名为它所绑定的资产——q.write(frame, "orders_augmented")，而不是 q.write(frame, "target")。下方示例中的 "target" 和 "source" 是占位符，不要照搬这些名称：它们可以正常编译，但会导致代码和绑定映射难以阅读，因为除了资产 ID 之外，没有任何信息能说明数据行会写入哪里。只要知道目标资产的名称，就应使用该名称。

每个绑定值可以是以下其中一种：

• 已注册的数据资产 — 数据资产 UUID，例如 {"<symbol>": "<data_asset_id>"}。系统会自动解析其连接和架构。

• 直接使用连接 — 直接绑定到通过 discover_connection_data 找到的表或文件，即使它不是已注册的数据资产也可以。格式为 "<connection_id>:<path>"，并且必须使用 discover_connection_data 返回的原样 path：

  {"<symbol>": "<connection_id>:/<SCHEMA>/<TABLE>"}
  

不要在绑定中放入列清单。对于数据库表，架构来自你在 q.source() 中声明的类型；对于文件，系统会自动获取架构（Excel 除外——见下文）。一如既往，只在 q.source() 中声明工作流实际使用的列。

基于文件的连接（Azure Blob Storage、S3、COS、SFTP 等）使用相同的 "<connection_id>:<path>" 格式，其中 path 是容器／存储桶和文件名：

• 仅支持从 Azure Blob Storage、S3 和 COS 源读取 Excel 文件。

• 系统会根据扩展名自动推断文件格式（.csv、.xlsx、.xls、.xlsm、.xlsb、.json、.parquet 等）。无需额外属性。

• 要指定 Excel 文件中的特定工作表，请将工作表名称作为路径末尾的一段，附加在文件名之后：

  {"<symbol>": "<connection_id>:/my-container/Report.xlsx/Sheet2"}
  

省略工作表路径段时，将读取第一个（默认）工作表。

文件源格式属性 [datastage]

对于分隔符分隔的文件，q.source() 接受可选关键字参数 "had_headers" 标志；Azure Blob Storage 源则接受另外 4 个属性。 如果用户提到了这些属性，或需要用它们修复运行时错误，请设置这些属性。

关键字参数	可接受的值	默认值
has_headers	True / False	True
field_delimiter	"comma"、"tab"、"colon" 或任意单个自定义字符	"comma"
quote_character	"double_quote"、"single_quote" 或 "none"	"none"
escape_character	"double_quote"、"single_quote"、"backslash" 或 "none"	"none"
row_delimiter	"new_line"、"carriage_return"、"carriage_return_line_feed" 或 "line_feed"	"new_line"
character_encoding	任意 IANA 编码字符串，例如 "UTF-8"、"ISO-8859-1"	连接器默认值

# Example: tab-separated file with UTF-8 encoding
sales = q.source(
    "sales_tsv",
    id="i64",
    name="string",
    amount="f64",
    has_headers=True,
    row_delimiter="new_line",
    field_delimiter="comma",
    quote_character="none",
    escape_character="backslash",
    character_encoding="utf-8",
)
q.name("azure_tsv_ingest")
q.write(sales, "sales_output", operation="overwrite")

解析绑定 — 优先使用数据资产

需要将符号绑定到数据源时，务必先尝试调用 list_data_assets，再遍历任何连接：

1. 调用 list_data_assets(project_id=<project>, entity_name="<table or file name>")。
2. 如果返回了匹配的资产，请使用其 UUID 作为绑定——{"<symbol>": "<data_asset_id>"}。编译器会自动解析连接和架构；无需调用 discover_connection_data。
3. 只有当 list_data_assets 未返回匹配项时（该表不是已注册的项目资产），才应回退到 discover_connection_data：遍历相关连接（连接 → 架构 → 表），并绑定 "<connection_id>:/<SCHEMA>/<TABLE>"。

这个顺序很重要：数据资产绑定包含更丰富的元数据，在编译时解析更快，也能避免不必要的连接遍历。只有当源没有对应的数据资产时，才将 discover_connection_data 作为最后手段。

本地参数 [datastage]

直接连接绑定路径中可以包含 #name# 形式的 DataStage 本地参数标记。作业运行时，DataStage 会将每个标记替换为该参数的当前值，因此无需重新编译，同一编译后的流程就可以读取或写入不同的表。

向 create_pyflow 传入 parameters，以声明每个标记及其默认值：

# DSL code — symbols are unchanged
orders = q.source("orders", id="i64", amount="f64")
q.name("parameterized_table_flow")
q.write(orders, "target", operation="insert")

# create_pyflow call
bindings = {
    "orders": "conn-id:/MYSCHEMA/#src_table#",
    "target": "conn-id:/MYSCHEMA/#tgt_table#",
}
parameters = {
    "src_table": "ORDERS_2024",    # default value
    "tgt_table": "ORDERS_ARCHIVE", # default value
}

规则：

• parameters 仅适用于 DataStage；如果设置的 engine= 是其他值，传入该参数会引发错误。
• 任何绑定路径中出现的每个 #token#，都必须在 parameters 中有对应的键，DataStage 宏除外（见下一节）。缺少非宏条目的情况会在编译时被拒绝。
• 如果没有合适的默认值，默认值可以是空字符串。
• 即使 parameters 中的键没有出现在任何绑定路径中，也仍会在流程中注册，并可通过 SDK 在阶段表达式中使用。
• 不要在数据资产 UUID 绑定中使用 #token#——标记仅在直接连接路径中有意义。

运行时覆盖——向 create_job_run 传入 runtime_parameters，无需重新编译：

create_job_run(
    job_ids=["<job_id>"],
    project_id="<project_id>",
    runtime_parameters={
        "local_parameters": {
            "src_table": "ORDERS_2025",
            "tgt_table": "ORDERS_ARCHIVE_2025",
        }
    }
)

参数集 [datastage]

参数集是一种项目级资产，可将多个命名参数归为一组。在绑定路径中，参数集参数使用点分隔标记引用：#setname.paramname#。参数集名称与参数名称之间的点，是区分参数集引用和本地参数（#name#，不含点）的标志。

与本地参数不同，参数集不在 parameters 中声明——create_pyflow 会自动检测绑定中的每个 #setname.paramname# 标记，验证项目中是否存在对应的参数集及其引用的每个参数，并将该参数集附加到编译后的流程。

**调用 create_pyflow 前，参数集必须已存在。**先使用 create_parameter_set 创建参数集，再使用 list_parameter_sets 确认名称。

**如果创建流程时使用了 #setname.paramname# 绑定标记，请勿在调用 create_pyflow 后调用 attach_parameter_set_to_flow。**编译器会在编译过程中附加参数集。之后调用 attach_parameter_set_to_flow 总是多余的——该工具会返回 status: already_attached，不执行任何操作。只有在为未使用引用该参数集的任何绑定标记而创建的流程添加参数集时，才调用 attach_parameter_set_to_flow。

# DSL code — unchanged from any other flow
orders = q.source("orders", id="i64", amount="f64")
q.name("env_paramset_flow")
q.write(orders, "target", operation="insert")

# create_pyflow call — no 'parameters' argument needed for the set
bindings = {
    "orders": "conn-id:/#EnvParams.SCHEMA#/#EnvParams.SRC_TABLE#",
    "target": "conn-id:/#EnvParams.SCHEMA#/ORDERS_ARCHIVE",
}
# EnvParams is a project parameter set with at least SCHEMA and SRC_TABLE parameters.
# No entry in 'parameters' is needed — the tool handles everything automatically.

同一流程中支持混用参数集和本地参数：

bindings = {
    "orders": "conn-id:/#EnvParams.SCHEMA#/#src_table#",
    "target": "conn-id:/#EnvParams.SCHEMA#/ORDERS_OUT",
}
parameters = {
    "src_table": "ORDERS_2024",   # local parameter — must be declared here
    # EnvParams is a parameter set — do NOT add its params here
}

运行时覆盖——通过 create_job_run 选择命名值集或覆盖个别参数，无需重新编译：

create_job_run(
    job_ids=["<job_id>"],
    project_id="<project_id>",
    runtime_parameters={
        "parameter_sets": [
            {
                "name": "EnvParams",
                "value_set": "prod",   # switch to the 'prod' value set
            }
        ]
    }
)

规则：

• 参数集标记（#setname.paramname#）仅适用于 DataStage——如果设置的 engine= 是其他值，使用此类标记会引发错误。
• 参数集名称和参数名称都必须是有效标识符（[A-Za-z_]\w*）。无效名称会在编译时被拒绝。
• 调用 create_pyflow 前，项目中必须存在该参数集。缺少参数集或参数名称拼写错误时，会在编译时被拒绝，并显示列出可用项的详细错误信息。
• 不要将参数集参数名称添加到 parameters 中。点分隔语法（#set.param#）用于让工具区分参数集参数和本地参数（#name#）。
• 单个流程中可以引用多个参数集——所有参数集都会经过验证并附加。
• 参数集标记可以与本地参数（#name#）在同一绑定路径中自由组合。

DataStage 宏 [datastage]

DataStage 宏是内置的全局参数，其值会在作业启动时由并行引擎自动解析——无需在 parameters 中添加任何条目。可以像使用本地参数一样，在直接连接绑定路径中使用宏（以 #…# 作为分隔符），但绝不能将其添加到 parameters 映射中。

# DSL code — unchanged
orders = q.source("orders", id="i64", amount="f64")
q.name("macro_demo_flow")
q.write(orders, "target", operation="insert")

# create_pyflow call — no entry in parameters for the macro
bindings = {
    "orders": "conn-id:/MYSCHEMA/#DSProjectName#_ORDERS",
    "target": "conn-id:/MYSCHEMA/#tgt_table#",
}
parameters = {
    "tgt_table": "ORDERS_ARCHIVE",  # local parameter — must be declared
    # Do NOT add DSProjectName here — it is a macro, not a local parameter
}

可用宏：

宏	运行时注入的值
#DSFlowName#	DataStage 流的名称
#DSHostName#	引擎层的主机名
#DSJobName#	作业名称
#DSJobStartDate#	作业开始日期（YYYY-MM-DD）
#DSJobStartTime#	作业开始时间（HH:MM:SS）
#DSJobStartTimestamp#	合并后的作业开始日期和时间
#DSJobWaveNo#	当前作业运行的波次（调用）编号
#DSProjectName#	DataStage 项目名称
#DSProjectDirectory#	引擎上项目目录的文件系统路径
#DSProjectId#	DataStage 项目的唯一 ID
#DSJobRunId#	当前作业运行的唯一 ID
#DSJobId#	作业的唯一 ID
#DSJobController#	作业控制器进程的主机名

规则：

• 宏仅适用于 DataStage——在任何其他引擎中都没有意义。
• 不要在 parameters 中声明宏名称。验证器会接受未声明的宏；如果尝试声明宏，则会报错（宏在运行时无论如何都会被忽略）。
• 同一路径中可以组合使用宏和本地参数标记，例如 "conn-id:/#DSProjectName#/#src_table#"。
• 宏名称区分大小写——请使用上表所示的准确大小写。

类型

i8  i16  i32  i64       signed integers
f32  f64                floating point
string  boolean         text, true/false
date  time  timestamp   temporal

Python 字面量会自动转换：int -> i64、float -> f64、str -> string、bool -> boolean。DSL 代码中绝不要写可空后缀（?）；后缀仅出现在目录元数据中。

q 命名空间

q 仅提供此处列出的方法。不要在 q 上调用其他任何方法——没有 q.lit()、q.sum()、q.avg()、q.max()、q.round()、q.row_number()、q.select()、q.datediff()，也没有任何自由聚合或标量函数。这些方法都属于 Expr，通过 q.col(name).method() 调用。

q.source(symbol, {"col": "type", ...}) -> Frame   # dict form; supports names with spaces/punctuation; at least one column must be provided
q.source(symbol, col="type", ...) -> Frame         # kwargs form; identifier-safe names; at least one column must be provided
q.name(name)                              # flow name; snake_case; exactly once — see Flow Naming below
q.output(frame, name)                     # register final output; required name; no other kwargs accepted
q.sink()                                  # register destination stage to discard incoming records
q.write(frame, symbol, operation="insert"|"overwrite"|"update"|"create")  # write final output to destination
q.col(name) -> Expr                       # column reference — use this, not frame['col'] or frame.col(name)
q.count_star() -> Expr                    # count-all `[datastage]`; use in .select() or .group_by().agg()
q.cast(value, type) -> Expr               # typed literal or expr cast; null: q.cast(None, "f64")
q.when(cond).then(val)...                 # see Conditional
q.concat(*exprs) -> Expr                  # string concat; 2+ args
q.date_diff(d1, d2) -> Expr               # day difference as i64 — note: date_diff, not datediff
q.strptime_time(expr, fmt) -> Expr        # string -> temporal; fmt is a strftime-style format
q.strftime(expr, fmt, tz?) -> Expr        # temporal -> string; tz is an IANA name

输出操作

q.output() 会将一个数据帧注册为流的最终输出，并在项目中创建一个包含该数据帧完整数据的文件数据资产：

q.output(frame, name="my_output")

name 为必填项。输出文件名为 {name}.csv，每次运行都会覆盖该文件。使用 q.sink(frame, name) 可实现相同效果；省略 q.sink() 中的 name 会生成不可预测的文件名——请使用 create_pyflow 响应中的 target_info[].target_path 查找该文件名。

若要在运行后查找输出资产，请调用 list_data_assets，并设置 entity_name={name}.csv。如果没有返回任何内容，则表示运行尚未完成。

read_data_preview 最多返回 100 行。资产包含数据帧的完整数据；该工具只显示其中的样本，并报告 row_count 和 truncated。当 truncated 为 true 时，不要将这些行呈现为完整结果，也不要用它们回答“有多少行”——实际行数是 poll_datastage_job 返回的 rows_written，即引擎自身记录的写入行数。

请预览 create_pyflow 的 target_info[].target_path 中指定的资产，而不是源表。如果读取输出资产失败，请如实说明，不要将输入表描述为结果。

写入操作

q.write() 使用绑定符号写入目录资产（由连接支持的表）：

q.write(frame, "target")                         # same as operation="insert"
q.write(frame, "target", operation="insert")     # append rows
q.write(frame, "target", operation="overwrite")  # replace the table's contents
q.write(frame, "target", operation="update")     # update existing rows
q.write(frame, "target", operation="create")     # create a new (non-existent) table from the frame's schema

• operation："insert" | "overwrite" | "update" | "create"。
• "create" 仅支持 DataStage 目标。
• "overwrite" 会在写入前清空表，因此重新运行工作流具有幂等性。当目标应仅包含本次运行的输出时使用它（仅限 DataStage）。
• 不支持 "upsert" 等操作；请勿用 insert 或 update 近似实现。
• "target" 符号的绑定方式与其他符号相同（参见 Symbols And Bindings）：可以使用已注册的数据资产 UUID，也可以直接使用连接 "<connection_id>:/<SCHEMA>/<TABLE>"，将数据直接写入由连接支持的表，而无需注册数据资产。
• 读取已写入的数据：运行工作流后，使用 read_data_preview 工具从目标连接读取数据。该工具最多返回 100 行；写入行数应以 poll_datastage_job 返回的 rows_written 为准，而不是预览结果的长度。

Create 操作行为

在 DataStage 上使用 operation="create" 时：

1. 表不存在：将根据帧的列推断架构并创建表。
2. 表已存在：不要对现有表使用 operation=create。
3. 绑定要求：必须使用直接连接绑定（"connection_id:/SCHEMA/TABLE"），而不是数据资产 ID，因为该表尚未在目录中注册。

示例：

# Create a new table from source data
source = q.source("input_table", id="i64", name="string", amount="f64")
q.name("create_new_table_flow")

# Use direct connection binding for non-existent table
q.write(source, "new_table", operation="create")

# In bindings:
bindings = {
    "input_table": "existing-asset-id-123",
    "new_table": "connection-id-456:/MYSCHEMA/NEW_TABLE"  # Direct connection binding
}

架构推断：

• 列名和类型取自帧的架构
• 不会自动创建主键、索引和约束
• 如需高级建表选项，请先手动创建表，再使用 operation="insert"

表达式方法

运算符返回的是 Expr，而不是 Python 布尔值。使用 &/|/~，不要使用 and/or/not。每个比较都要加括号：(q.col("a") > 1) & (q.col("b") < 2)。

==  !=  >  <  >=  <=    comparison -> boolean
+  -  *  /              arithmetic
&  |  ~                 and / or / not

.alias(name)                              # snake_case
.cast(type)                               # q.col("x").cast("i32") — use for rounding/type coercion; no .round() method
.precision(n)                             # numeric precision hint for output columns; int only
.scale(n)                                 # numeric scale hint for output columns; int only
.sum()                                    # aggregate; both engines
.mean()/.avg() .count() .min() .max()     # aggregates; `[datastage]` only
.rank() .dense_rank() .row_number()       # window functions; `[datastage]` only; see Partitioning section
.is_in(v1, v2, ...)                       # or .is_in([v1, v2])
.is_null() .is_not_null()                 # null checks -> boolean;
.asc() .desc()                            # sort direction only
.nulls_first() .nulls_last()              # nulls position in sort

没有 .round() 方法。 若要减少小数位数，请转换为精度较低的类型：q.col("x").cast("f32")，或者对输出列使用 .precision(n).scale(n)。

Expr 没有 .fillna() 或 .coalesce() 方法。 若要替换空值，请使用条件表达式：q.when(q.col("x").is_null()).then(0).otherwise(q.col("x"))。对于两个参数的合并模式，也可以同样链式调用 .when().then().otherwise()。

列引用本身没有 .contains() 方法。 字符串谓词位于 .str 访问器下：q.col("x").str.contains("pattern")。

聚合函数和窗口函数必须在 Expr 对象上调用（q.col(...).sum()、q.col(...).row_number()），不能直接在 q 或 Frame 上调用。它们仅可在 .select()、.group_by().agg() 或 .partition_by().select() 中使用。

.precision(n) 和 .scale(n) 可按任意顺序链式调用，且仅适用于 f64/numeric/decimal/double 列。如果设置了 .precision(n) 但省略 .scale(n)，当 n > 20 时，默认 scale 为 10；否则为 0。

它们必须放在直接传入 q.output() 或 q.write() 的最后一个 .with_columns() 或 .select() 上。 放在任何更早的（中间）节点上的注解都会被忽略；发生这种情况时，create_pyflow 响应中会返回警告。

# ✓ CORRECT — annotations on the node fed directly into q.output()
q.output(
    frame.with_columns(
        q.col("amount").precision(18).scale(4),
        q.col("rate").precision(10).scale(6),
    ),
    name="result",
)

# ✓ CORRECT — annotations on the node fed directly into q.write()
q.write(
    frame.with_columns(
        q.col("amount").precision(18).scale(4),
    ),
    "target_table",
    operation="insert",
)

# ✗ WRONG — annotations set mid-pipeline; the downstream .select() strips them
intermediate = frame.with_columns(
    q.col("amount").precision(18).scale(4),  # ignored — not the last node
)
q.output(
    intermediate.select("amount", "rate"),   # annotations lost here
    name="result",
)

条件表达式

q.when(cond).then(val)                                      # else is NULL
q.when(cond).then(val).otherwise(else_val)
q.when(c1).then(v1).when(c2).then(v2).otherwise(else_val)   # multi-branch

.str 访问器

字符串方法通过 .str.<method>() 访问，不能直接在列引用上调用。q.col("x").contains(...) 是错误的；应使用 q.col("x").str.contains(...)。

.str.upper()
.str.contains(s)       .str.starts_with(p)   .str.ends_with(s)
.str.like(pattern)     .str.replace(old, new)
.str.trim(chars?)      .str.rtrim(chars?)
.str.substring(start_1based, length?)

最佳实践： 除非有明确要求保留空格或用户明确提出要求，否则应对最终输出结果中的字符串列应用 .str.trim()，以删除首尾空白。默认情况下，优先采用干净且已去除首尾空白的最终输出。

Frame 方法

.filter(expr) -> Frame                    # boolean expr; no aggregates inside
.select(*exprs) -> Frame                  # bare strings become col(name); mixing plain refs with aggregates triggers implicit group-by
.with_columns(*exprs) -> Frame            # keep all input cols + add/replace; no aggregates inside; all expressions must use .alias(); no keyword-argument form
.sort(*col_refs) -> Frame                 # column refs only; bare strings sort asc; use .asc()/.desc()/.nulls_first()/.nulls_last()
                                          # nulls position defaults to nulls_first when not specified
.head(count) -> Frame                     # use .fetch(count, offset) when offset needed; not .limit()
.unique(*subset) -> Frame                 # empty subset dedupes on all columns; not .distinct() or .drop_duplicates()
.union(other) -> Frame                    # set-semantics dedup
.intersect(other) -> Frame
.partition_by(*col_refs, order_by=?) -> _PartitionBuilder  # `[datastage]` only; see Partitioning section

Frame 方法名称不可替换为同义词。 以下写法都会导致编译错误：.order_by()（使用 .sort()）、.limit()（使用 .head()）、.distinct() / .drop_duplicates()（使用 .unique()）、.left_join() / .right_join()（使用 .join(..., how="left")）、.groupby()（使用带下划线的 .group_by()）、单数形式的 .with_column()（使用复数形式的 .with_columns()）、直接在 Frame 上调用 .agg()（使用 .group_by(...).agg(...)）。

通过 q.col('name') 引用列，而不是使用下标。 不支持 frame['col']，请在表达式中使用 q.col('col')。

with_columns() 接受位置参数 Expr，每个参数都要使用 .alias()，而非关键字参数。 frame.with_columns(new_col=q.col("x") + 1) 是错误写法。请写成 frame.with_columns((q.col("x") + 1).alias("new_col"))。

聚合函数只能出现在 .select() 或 .group_by().agg() 中。要基于聚合值进行筛选，请先聚合，再调用 .filter(...)。

不支持分析型窗口函数。 有关 DataStage 窗口函数，请参阅下方的分区部分。

DataStage 上的滚动 / 范围窗口聚合（例如“对每个锚点行，汇总最近 27 天内的贡献值”）不是内置操作。请通过 .cross() → .filter(q.date_diff(...) <= N) → .group_by().agg() 实现。示例部分提供了相应方案。

Join [datastage]

a.join(b, on=, how="inner", suffix="_right") -> Frame
a.join(b, left_on=, right_on=, how="inner", suffix="_right") -> Frame
a.cross(b, suffix="_right") -> Frame

• how：inner | left | right | outer | cross。
• on=（同名键）：右侧键列会被删除。键必须是字符串（列名），不能是表达式。键名不同时，请使用 left_on= / right_on=。
• left_on / right_on（不同名键）：保留两侧的键列。
• 重复的右侧非键列会添加 suffix（发生冲突时会叠加：_right_right）。使用 q.col("x_right").alias(...) 重命名。
• join() 只能接受一个位置参数（右侧 Frame），以及关键字参数。 a.join(b, q.col("id") == q.col("id2")) 是错误写法。请通过 on=、left_on= 或 right_on= 传入连接条件。如果键列名称不同，请使用 left_on="id", right_on="id2"。
• 传入 left_on= / right_on= 的列必须存在于对应一侧——left_on 必须指定左侧 Frame 中的列；right_on 必须指定右侧 Frame 中的列。如果出现“未找到列”错误，请对照源架构核实列名。
• 非等值连接和范围连接（不等式谓词，例如 a.date <= b.date、日期距离阈值或 id != id）不能使用 .join()，因为它只支持等值连接。请使用 .cross() 生成笛卡尔积，再通过 .filter() 筛选不等式条件。

分区 [datastage]

.partition_by(*col_refs, order_by=?) -> _PartitionBuilder

分区用于配置数据的分布和排序方式，以供 rank()、dense_rank() 和 row_number() 等窗口函数使用。.partition_by() 后只能链式调用 .select()——_PartitionBuilder 不支持 .with_columns()。窗口函数（.rank()、.dense_rank()、.row_number()）是 Expr 方法，应在 .select() 中调用，而不是作为 q 上的独立函数。必须至少提供一个分区键；如果要使用不分区的全局窗口，请使用常量列（例如，通过 q.cast(1, "i32").alias("_part") 添加字面量列）。

# Rank products by sales within each category
result = (
    sales
    .partition_by("category", order_by="amount")
    .select(
        q.col("category"),
        q.col("product"),
        q.col("amount"),
        q.col("amount").rank().alias("sales_rank")
    )
)

# Multiple partition keys and order columns
result = (
    sales
    .partition_by("region", "category", order_by=[("amount", "desc"), "date"])
    .select(
        q.col("region"),
        q.col("category"),
        q.col("product"),
        q.col("amount").row_number().alias("row_num")
    )
)

• 分区键：用于分区的列名（将数据分配到不同组）
• order_by：可选。在每个分区内排序的列。可以是：
  • 单列：order_by="amount"
  • 列表：order_by=["amount", "date"]
  • 包含方向元组的列表：order_by=[("amount", "desc"), ("date", "asc")]
• 窗口函数：.rank()、.dense_rank()、.row_number() 在分区边界内计算
• 常规函数：标量函数（如 .str.upper()、q.strftime()、算术运算等）可在同一个 .select() 中正常使用——分区仅影响窗口函数
• 分区为可选项：窗口函数可以不使用 .partition_by()（此时整个数据集作为一个分区处理），不过为了性能和正确性，建议使用分区

Group-By 与 Select 中的聚合 [datastage]

.group_by(*col_refs).agg(*measures) -> Frame     # .alias() every measure

• .group_by() 至少需要一个列参数。如果要进行不带分组键的全局聚合，请改用带聚合表达式的 .select()：t.select(q.col("x").sum().alias("total"))。
• .agg() 接受位置参数 Expr——不接受关键字参数。agg(total=q.col("x").sum()) 是错误写法。应写为 agg(q.col("x").sum().alias("total"))。
• 分组列必须是简单的列引用（裸列名字符串或 q.col("name")），不能是计算表达式。若要按派生值分组，请先使用 .with_columns() 将其物化，再将该列名传给 .group_by()。

在 .select() 中混用普通列引用和聚合表达式时，普通列引用会成为隐式分组键：

t.select(q.col("x").sum().alias("total"))               # global aggregate
t.select("status", q.col("x").count().alias("n"))       # grouped by status

对于计算得出的分组键，请先使用 .with_columns() 将其物化，再按该列名分组。

流程命名

传给 q.name() 的名称会直接用作流程名称。设置 replace_flow_id 时，目标流程会保留其现有名称，并忽略 q.name()。

重试时，请通过 replace_flow_id 复用现有流程。

示例

DataStage 聚合：类型化字面量筛选、在 select 中聚合，然后筛选聚合列：

orders = q.source("orders", {"region": "string", "amount": "f64", "orderdate": "date"})
q.name("top_regions_2024")
q.output(
    orders
    .filter(q.col("orderdate") >= q.cast("2024-01-01", "date"))
    .select("region", q.col("amount").sum().alias("revenue"), q.count_star().alias("n_orders"))
    .filter(q.col("n_orders") > 100)
    .sort(q.col("revenue").desc())
    .head(10),
    name="top_regions_output"
)

DataStage 链式连接；右侧重复的名称会添加 "_right"，并通过 .alias() 重命名：

customer = q.source("customer", {"custkey": "i64", "name": "string", "nationkey": "i64"})
nation = q.source("nation", {"nationkey": "i64", "name": "string", "regionkey": "i64"})
region = q.source("region", {"regionkey": "i64", "name": "string"})
q.name("customer_geography")
q.output(
    customer.join(nation, on="nationkey")
    .select("custkey", "name", "regionkey", q.col("name_right").alias("nation"))
    .join(region, on="regionkey")
    .select("custkey", "name", "nation", q.col("name_right").alias("region")),
    name="customer_geography_output"
)

DataStage 分区窗口函数：

sales = q.source("sales", {"category": "string", "product": "string", "amount": "f64", "date": "date"})
q.name("ranked_products")
q.output(
    sales
    .partition_by("category", order_by=[("amount", "desc")])
    .select(
        q.col("category"),
        q.col("product"),
        q.col("amount"),
        q.col("date").strftime("%Y-%m").alias("month"),  # regular function works fine
        q.col("amount").rank().alias("sales_rank"),
        q.col("amount").row_number().alias("row_num")
    )
    .filter(q.col("sales_rank") <= 10),  # top 10 per category
    name="top_products_output"
)

DataStage 滚动／范围窗口聚合——交叉连接模式 [datastage]：

# For each anchor row, sum all contributions that fall within 27 days before it.
# The Join stage is equi-only, so a range predicate goes through .cross() + .filter().
anchors = q.source("anchors", {"id": "i64", "anchor_date": "date", "region": "string"})
contribs = q.source("contribs", {"anchor_id": "i64", "contrib_date": "date", "amount": "f64"})
q.name("rolling_27d_sum")
q.output(
    anchors
    .cross(contribs, suffix="_c")                               # full cartesian product
    .filter(
        (q.col("id") == q.col("anchor_id_c")) &                # equi predicate first (cheap filter)
        (q.date_diff(q.col("anchor_date"), q.col("contrib_date_c")) >= 0) &
        (q.date_diff(q.col("anchor_date"), q.col("contrib_date_c")) <= 27)
    )
    .group_by("id", "anchor_date", "region")
    .agg(q.col("amount_c").sum().alias("rolling_sum")),
    name="rolling_output"
)

DataStage Azure Blob Storage 文件读取，配置分隔符属性 [datastage]：

sales = q.source(
    "sales_tsv",
    id="i64",
    name="string",
    amount="f64",
    has_headers=True,
    row_delimiter="new_line",
    field_delimiter="comma",
    quote_character="none",
    escape_character="backslash",
    character_encoding="utf-8",
)
q.name("azure_tsv_ingest")
q.write(sales, "sales_output", operation="overwrite")

要点：

• .cross() 是通过不等式或日期距离谓词进行连接的唯一方式；.join() 仅支持等值连接。
• 在 .filter() 中先放置等值谓词（id == anchor_id），这样引擎会在计算范围条件之前丢弃不匹配的组合。
• 当 later >= earlier 时，q.date_diff(later, earlier) 会返回正数 i64。如果符号相反，请交换参数。
• .filter() 之后，将结果视为普通数据帧：可使用 .group_by().agg()、.select()、继续 .join() 等。