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 でブートストラップしてから、pyflow では表現できないもの（式、プロパティ、追加のカスタム／Buildop／Java ステージなど）を生成された SDK に精密に編集して加えます。pyflow は基盤を確実に組み立てるので、残りは後から差し込みます。
• 既存の構造に対する式またはプロパティのみで、新しい配線がない場合 → 取得した SDK をそのまま編集します。
• コアとなる形状を忠実に表現する pyflow の形式がなく、ブートストラップすると追加ではなく再配線が必要な骨組みが生成される場合 → pyflow で作成可能な最も近い形状をブートストラップしてから、update_datastage_flow を使って生成された SDK の形状を変更します。ステージがないことだけでは「pyflow の形式がない」とは見なされません。ステージは後から（上記の手順で）差し込めるためです。

コードの構造

ランタイムによって q が提供されます。インポートしたり、インスタンス化したりしないでください。すべてのフローは次のとおりです。

1. q.source() でソースを宣言します。アセットのメタデータにある正確な名前と型を使い、参照する列のみを列挙してください。少なくとも1列を指定する必要があります。列を指定しない q.source("sym") はコンパイルエラーになります。
2. q.name("<snake_case_name>") を必ず1回だけ呼び出します。
3. シンクは必ず1つだけにします。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 Namespace セクションに記載されたメソッドのみです。.lit()、.sum()、.avg()、.max()、.round()、.row_number()、.select()、またはそこに記載されていないその他のメソッドはありません。集計、ウィンドウ関数、スカラー演算は q 自体ではなく、q.col(...) が返す Expr オブジェクトにあります。

エンジンのターゲット

呼び出し元は create_pyflow(engine=...) にターゲットエンジンを渡します。コード内で宣言しないでください。エンジンによって、許可される Frame 操作が決まります。

Op	DataStage	Jetstream
q.source() / q.debug_source()	任意の数	1つのみ
q.output() / q.sink() / q.write()	はい	はい
.filter()、.sort()	はい	いいえ
.lookup()	いいえ	はい
.tumble() / .slide().agg()	いいえ	はい
.select() / .with_columns()	はい	いいえ
.head() / .fetch()、.unique()	はい	いいえ
.union()、.intersect()	はい	いいえ
.group_by().agg()	はい	いいえ
.join()、.cross()	はい	いいえ

Jetstream のフローは、単一の直線的なチェーンにする必要があります。

q.source() | q.debug_source() -> [.lookup()]* -> q.write()

filter、sink などの他の操作は開発中であり、Jetstream ではまだサポートされていません。

シンボルとバインディング

q.source()、.lookup()、.write() に渡す文字列は、ローカルのシンボルです。呼び出し元は create_pyflow(bindings=...) を使って、各シンボルをデータソースにバインドします。シンボル名はカタログ名と一致している必要はありません。使用するすべてのシンボルをバインドする必要があります。

各シンボルには、バインド先のアセットにちなんだ名前を付けてください — q.write(frame, "orders_augmented") のようにします。q.write(frame, "target") ではありません。以下の例にある "target" と "source" はコピーして使う名前ではなく、プレースホルダーです。コンパイルは問題なく行えますが、行き先を示す情報がアセット ID しかないため、コードとバインディングマップが読みづらくなります。可能な限り、出力先アセット自体の名前を使ってください。

各バインディング値は次のいずれかです。

• 登録済みデータアセット — データアセットの UUID。例：{"<symbol>": "<data_asset_id>"}。接続とスキーマは自動的に解決されます。

• 接続を直接使用 — discover_connection_data で見つけたテーブルまたはファイルに直接バインドします。登録済みデータアセットでない場合も使用できます。discover_connection_data が返した path をそのまま使い、"<connection_id>:<path>" の形式で指定します。

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

バインディングに列リストを含めないでください。データベーステーブルの場合、スキーマは型を指定した q.source() 宣言から取得されます。ファイルの場合は自動的に取得されます（Excel を除く。詳細は後述）。通常どおり、q.source() ではフローが実際に使用する列だけを宣言してください。

ファイルベースの接続（Azure Blob Storage、S3、COS、SFTP など）も、path にコンテナー／バケットとファイル名を指定する同じ "<connection_id>:<path>" 形式を使います。

• Excel の読み取りに対応しているのは、Azure Blob Storage、S3、COS のソースのみです。

• ファイル形式は拡張子（.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	"UTF-8"、"ISO-8859-1" など、任意の IANA エンコーディング文字列	コネクターのデフォルト

# 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="jetstream" とともに渡すとエラーになります。
• いずれかのバインディングパスに含まれるすべての #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="jetstream" で使用するとエラーになります。
• セット名とパラメーター名は、どちらも有効な識別子（[A-Za-z_]\w*）である必要があります。無効な名前はコンパイル時に拒否されます。
• create_pyflow を呼び出す前に、パラメーターセットがプロジェクトに存在している 必要があります。セットが見つからない場合や、パラメーター名に誤りがある場合、利用可能な項目を示す説明的なエラーとともにコンパイル時に拒否されます。
• パラメーターセットのパラメーター名を parameters に追加しないでください。ドット区切りの構文（#set.param#）によって、ツールはそれらをローカルパラメーター（#name#）と区別します。
• 1 つのフローで複数のパラメーターセットを参照できます。すべてのセットが検証され、関連付けられます。
• パラメーターセットのトークンは、同じバインディングパス内でローカルパラメーター（#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 専用です。StreamSets では意味を持ちません。
• マクロ名を 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 コードでは nullable サフィックス（?）を記述しないでください。サフィックスはカタログのメタデータでのみ使用します。

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.source(symbol, col="type", ..., schema_metadata={...}) -> Frame  # with schema metadata (Kafka/StreamSets)
q.debug_source()                          # debug source stage for development: Dev Raw Data Source
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

シンク操作／Trash 送信先ステージの処理手順 [jetstream]

Trash 送信先は、受信したすべてのレコードを破棄するシンクです。そのため、スキーマは不要で、データアセットを参照する必要もありません。

Pyflow コードの例：

source_data = q.source("pg_table", {"id":"i32", "name": "string"})

# Name the flow
q.name("pg_discard")

# Sink to Trash
q.sink(source_data)

ソーススキーマのメタデータ [jetstream]

StreamSets 上の Kafka ソースでは、必要に応じてスキーマレジストリのメタデータを指定できます。

q.source(symbol, col="type", ..., schema_metadata={"subject": "...", "format": "..."})

パラメーター：

• schema_metadata：スキーマレジストリ情報を含む任意の dict：
  • "subject"：スキーマレジストリのサブジェクト名（format が AVRO の場合のみ使用。既定値は {symbol}-value）
  • "format"：データのシリアル化形式。"AVRO"、"JSON"、または "PROTOBUF"（既定値は "AVRO"）

動作：

• スキーマレジストリへの問い合わせは行われません。 すべての値は明示的に指定するか、既定値が使用されます。
• schema_subject は format が AVRO の場合のみ使用されます。 JSON／PROTOBUF の場合、subject は無視されます。
• schema_metadata が指定されていない場合：format の既定値は "AVRO"、subject の既定値は {topic}-value
• "subject" のみが指定されている場合：format の既定値は "AVRO"、subject には指定値が使用されます
• "format" のみが指定されている場合：指定された format が使用されます。AVRO の場合、subject の既定値は {topic}-value

例：

# AVRO with custom subject (subject is used)
orders = q.source("orders",
                  order_id="i64",
                  amount="f64",
                  schema_metadata={
                      "subject": "orders-value-v2",
                      "format": "AVRO"
                  })

# JSON format (subject is ignored even if provided)
orders = q.source("orders",
                  order_id="i64",
                  amount="f64",
                  schema_metadata={"format": "JSON"})

# Subject only (format defaults to AVRO, subject is used)
orders = q.source("orders",
                  order_id="i64",
                  amount="f64",
                  schema_metadata={"subject": "orders-value-v2"})

# No metadata (format defaults to AVRO, subject defaults to "orders-value")
orders = q.source("orders", order_id="i64", amount="f64")

ユースケース：

• ネットワークポリシーにより、コンパイル中のスキーマレジストリへのアクセスが禁止されている
• {topic}-value の規則に従わない、カスタムの AVRO スキーマ subject 命名
• データ形式（AVRO、JSON、PROTOBUF）を明示的に制御する

出力操作

q.output()はフレームをフローの最終出力として登録し、フレームの全データを含むデータファイルアセットをプロジェクト内に作成します。

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

nameは必須です。出力ファイル名は{name}.csvで、実行のたびに上書きされます。同じ効果を得るにはq.sink(frame, name)を使用します。q.sink()でnameを省略すると予測できないファイル名が付けられます。ファイル名を確認するには、create_pyflowのレスポンスにあるtarget_info[].target_pathを使用してください。

実行後に出力アセットを見つけるには、entity_name={name}.csvを指定してlist_data_assetsを呼び出します。何も返されない場合は、まだ実行が完了していません。

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"シンボルはほかのシンボルと同様にバインドされます（シンボルとバインディングを参照）。登録済みデータアセットのUUID、またはデータアセットを登録せずに接続に紐付いたテーブルへ直接書き込む場合は、"<connection_id>:/<SCHEMA>/<TABLE>"を使用します。
• 書き込んだデータの読み取り：フローの実行後、read_data_previewツールを使用して接続先からデータを読み取ります。返されるのは最大100行です。書き込まれた件数はプレビューの行数ではなく、poll_datastage_jobのrows_writtenです。

Create操作の動作

DataStageでoperation="create"を使用する場合：

1. テーブルが存在しない場合：フレームの列から推論されたスキーマでテーブルが作成されます。
2. テーブルがすでに存在する場合：既存のテーブルにoperation=createを使用しないでください。
3. バインディング要件：テーブルはまだカタログに存在しないため、データアセットIDではなく、直接接続バインディング（"connection_id:/SCHEMA/TABLE"）を使用する必要があります。

例：

# 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"を使用してください。

式のメソッド

演算子は Python の bool ではなく、Expr を返します。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() メソッドはありません。 null を置き換えるには、条件式を使用します：q.when(q.col("x").is_null()).then(0).otherwise(q.col("x"))。2 引数の coalesce パターンには、同じように .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 ならデフォルトのスケールは 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() は、キーワード引数ではなく、.alias() を持つ Expr 引数を位置引数として受け取ります。 frame.with_columns(new_col=q.col("x") + 1) は誤りです。frame.with_columns((q.col("x") + 1).alias("new_col")) と記述してください。

集計関数は .select() または .group_by().agg() の内部でのみ使用できます。集計値でフィルターする場合は、先に集計し、その後 .filter(...) を使用してください。

分析ウィンドウ関数はありません。 StreamSets の時間ウィンドウ集計には .tumble()／.slide() を使用してください。DataStage のウィンドウ関数については、以下の「パーティショニング」セクションを参照してください。

DataStage のローリング／範囲ウィンドウ集計（例：「各アンカー行について、直近 27 日間の寄与の合計」）は、組み込み操作ではありません。.cross() → .filter(q.date_diff(...) <= N) → .group_by().agg() として記述してください。「例」セクションのレシピを参照してください。

[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() が受け取る位置引数は、右側のフレームの1つだけです。ほかの引数はキーワード引数で指定してください。 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 には左側のフレームの列、right_on には右側のフレームの列を指定してください。「column not found」エラーが発生した場合は、ソースのスキーマで列名を確認してください。
• 非等価結合と範囲結合（a.date <= b.date のような不等式述語、日付間隔のしきい値、id != id など）には、等価結合のみ対応する .join() は使用できません。.cross() で直積を生成してから、不等式条件を .filter() で指定してください。

[jetstream] のルックアップ

m.lookup(symbol, {col: type, ...}, on=, suffix="_right") -> Frame
m.lookup(symbol, {col: type, ...}, left_on=, right_on=, suffix="_right") -> Frame
m.lookup(symbol, col=type, ..., on=) -> Frame                  # kwargs form

• 参照シンボルからインラインで読み込んだ列を使って m を拡張します。参照を別途 q.source() で宣言しないでください。
• 参照列には、参照ステージの全スキーマを指定します。on= / right_on= のキー列と、取り込む列を含めてください。
• セマンティクスは左結合に似ています。一致しない行は参照列が NULL のまま保持されます。最初の一致のみが使用されます。how= はありません。
• キーとサフィックスのルールは、上記の .join() と同じです。

ウィンドウ集計 [jetstream]

m.tumble(length, group_by=?, tz=?, on=?).agg(*measures) -> Frame
m.slide(length, group_by=?, tz=?, on=?).agg(*measures) -> Frame

• length: <number><unit>。unit は s | m | h | d です（例："30s"、"15m"、"1h"）。
• group_by: 文字列または列名のリストです（Jetstream では必須）。グローバルウィンドウを使用する場合は、定数列（例：q.cast(1, "i32").alias("_global")）を追加して、それをグループ化します。
• tz: IANA タイムゾーン。
• on: イベント時刻列です（Jetstream では必須）。データ内のイベントタイムスタンプを Unix ミリ秒で含むフィールドである必要があります（timestamp ではなく i64 型を使用してください）。
• 出力列：[*group_by, window_start, window_end, *measure_aliases]。window_start / window_end は timestamp です。
• メジャー：初期の StreamSets ウィンドウ処理コンパイラーがサポートするのは .sum() のみです。各メジャーに .alias() を指定する必要があります。ネストには対応していません。

重要： Jetstream のウィンドウ処理には、イベント時刻処理が必要です。タイムスタンプを Unix ミリ秒（Long 値）で含むフィールド名を on パラメーターで指定する必要があります。処理時刻ウィンドウはサポートされていません。 pyflow コンパイラーは、on フィールドをソースステージに自動的に引き継ぎます。コンパイラーが予約しており、指定するとエラーになるため、ソースの configs={} で watermarkStrategy や timestampRecordField を渡さないでください。

[datastage] によるパーティション分割

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

パーティション分割では、rank()、dense_rank()、row_number() などのウィンドウ関数で、データをどのように分配して順序付けするかを設定します。.partition_by() の後には .select() のみを連結してください。_PartitionBuilder では .with_columns() は使用できません。ウィンドウ関数（.rank()、.dense_rank()、.row_number()）は .select() 内で呼び出す Expr メソッドであり、q に対する単独の関数ではありません。パーティションキーは少なくとも1つ指定する必要があります。パーティション分割を行わないグローバルウィンドウでは、定数列を使用してください（例：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() なしでも使用でき、データセット全体を1つのパーティションとして処理します。ただし、パフォーマンスと正確性のため、パーティション分割を推奨します

グループ化と集計を含む Select [datastage]

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

• .group_by() には、少なくとも1つの列引数が必要です。グループ化キーなしで全体を集計する場合は、代わりに集計式を指定して .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"
)

StreamSets のルックアップ：参照スキーマはインラインで宣言し、別途 q.source() で宣言する必要はありません：

orders = q.source("orders", {"cust_id": "i64", "amount": "f64"})
q.name("orders_with_customer")
q.output(
    orders
    .filter(q.col("amount") > 0)
    .lookup("customer", {"cust_id": "i64", "name": "string"}, on="cust_id"),
    name="orders_with_customer_output"
)

イベント時刻列を使った StreamSets のウィンドウ集計：

events = q.source("events", {"region": "string", "amount": "f64", "ts": "i64"})  # ts as Unix milliseconds
q.name("revenue_15m")
q.output(
    events
    .tumble("15m", group_by="region", on="ts")
    .agg(q.col("amount").sum().alias("revenue")),
    name="revenue_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）を先に置いてください。これにより、範囲条件の評価前に、エンジンが一致しないペアを除外できます。
• q.date_diff(later, earlier) は、later >= earlier の場合に正の i64 を返します。符号が逆になる場合は引数を入れ替えてください。
• .filter() の後は、結果を通常のフレームとして扱います。.group_by().agg()、.select()、さらに .join()` などを使用できます。

ステージ設定 [jetstream]

オペレーションでは、基盤となるステージにエンジン固有の設定を渡すための、省略可能な configs パラメーターがサポートされています。設定は StreamSets のフローにのみ適用され、DataStage では無視されます。

q.source(symbol, col="type", ..., configs={"key": "value"})
q.debug_source(configs={"key": "value"})
q.write(frame, symbol, operation="insert", configs={"key": "value"})
frame.filter(expr, configs={"key": "value"})
frame.select(..., configs={"key": "value"})
frame.with_columns(..., configs={"key": "value"})
frame.sort(..., configs={"key": "value"})
frame.lookup(symbol, {...}, on=, configs={"key": "value"})
frame.tumble(length, configs={"key": "value"}).agg(...)
frame.slide(length, configs={"key": "value"}).agg(...)

ルール：

• configs は、文字列キーと JSON シリアライズ可能な値（str、int、float、bool、None、list、dict）を持つ dict である必要があります
• 予約済みのキーは使用できません：stageType、instanceName、stageName、uiInfo、inputLanes、outputLanes、eventLanes、services、configuration
• 標準プロパティの設定後、設定は StreamSets のステージ設定にマージされます
• 無効な設定がある場合、コンパイル時に ConfigValidationError が発生します

例：

# Source with batch size config
orders = q.source("orders_kafka",
                  order_id="i64",
                  amount="f64",
                  configs={"batchSize": 1000, "maxWaitTime": 5000})

# Debug source with rawData and dataFormat config
orders = q.debug_source(configs={"rawData":"id,total\n1,1.1\n2,2.2", "dataFormat":"CSV"})

# Write with custom buffer settings
q.write(result, "target_db",
        operation="insert",
        configs={"batchSize": 500, "connectionTimeout": 30})

# Filter with processing hints
filtered = orders.filter(
    q.col("amount") > 100,
    configs={"enableMetrics": True}
)

# Lookup with cache configuration
enriched = orders.lookup(
    "customers",
    {"cust_id": "i64", "name": "string"},
    on="cust_id",
    configs={"cacheSize": 10000, "cacheTTL": 3600}
)

# Window with custom watermark (ts must be i64 Unix milliseconds)
windowed = events.tumble(
    "15m",
    group_by="region",
    on="ts",  # ts field must contain Unix milliseconds
    configs={"allowedLateness": "5m"}
).agg(q.col("amount").sum().alias("revenue"))