ユーザー定義演算子 YAML リファレンス

Lakeflow Designer のユーザー定義演算子は YAML で定義されます。 すべての演算子の型 (uc-udf、 uc-udtf、および python-run-function) では、JSON スキーマ形式を使用して構成フィールドを定義する user-defined-operator-v0.1.0 スキーマが使用されます。

ユーザー定義演算子を作成する方法については、 Lakeflow Designer のユーザー定義演算子を参照してください。

ルート プロパティ

すべての演算子 YAML ファイルは、演算子を識別してその動作を定義するルート プロパティのセットで始まります。 次の例は、一般的な構造を示しています。

schema: user-defined-operator-v0.1.0
type: python-run-function
name: My Operator
id: my_operator
version: '1.0.0'
description: >
  What this operator does.
  Can be multiple lines.
config:
  type: object
  properties:
    my_field:
      type: string
      title: My Field
      description: Help text
ports:
  input:
    - name: data
      title: Input Data
  output:
    - name: out
      title: Output
run_function:
  type: inline
  code: |
    def run(config, inputs, spark):
        return {"out": inputs["data"]}
environment:
  environment_version: '4'
  dependencies:
    - 'pandas>=2.0'
プロパティ タイプ 必須 説明
schema 文字列 Yes スキーマ識別子。 user-defined-operator-v0.1.0である必要があります。
type 文字列 Yes 演算子の種類: uc-udf、 uc-udtf、または python-run-function。
name 文字列 Yes 演算子の表示名。 Lakeflow Designer UI に合わせて短くします。 最小長は 1 文字です。
id 文字列 Yes 演算子型の一意識別子。 最小長は 1 文字です。 名前空間 ( finance. や ml.など) を使用して演算子を分類することを検討してください。
description 文字列 Yes 演算子の動作の詳細な説明。 UI のユーザーに表示されます。 YAML の複数行構文 (>) を使用して、より長い説明を行います。
config オブジェクト Yes 構成フィールドを定義する JSON スキーマ オブジェクト。 構成を参照してください。
ports オブジェクト いいえ 入力ポートと出力ポートの定義。 「ポート」を参照してください。
version 文字列 Yes バージョン文字列 (たとえば、 "1.0.0")。 これを使用して、独自のオペレーターのリリースを追跡します。
run_function オブジェクト いいえ python-run-function 演算子のインライン Python コード。 run_functionを参照してください。
environment オブジェクト いいえ 依存関係を含む環境の構成をPythonします。 environmentを参照してください。

ポート

ポートは、オペレーターがパイプライン内の他の演算子に接続する方法を定義します。 ports オブジェクトには、input配列とoutput配列が含まれています。

ports:
  input:
    - name: input_data
      title: Input Data
      mime: application/vnd.databricks.dataframe
      allowMultiple: true
      required: true
  output:
    - name: out
      title: Output
プロパティ タイプ 必須 説明
name 文字列 Yes ポートの一意識別子。 接続と構成参照で使用されます。
title 文字列 いいえ 人間が判読できるラベルが UI に表示されます。
mime 文字列 いいえ ポート データの MIME の種類。 たとえば、「 application/vnd.databricks.dataframe 」のように入力します。
allowMultiple boolean いいえ true場合、ポートは複数の着信接続を受け入れます。 既定値は false で、ポートは単一の接続を受け入れ、新しいソースを配線すると既存の接続が置き換えられます。
required boolean いいえ false場合、ポートは省略可能です。 既定値: true。

文書化されたポート プロパティのみが受け入れられます。 不明なキー (レガシ label フィールドなど) は、スキーマの検証によって拒否されます。

ポートの例

入力ポートと出力ポートを含む UDF:

ports:
  input:
    - name: in
      title: Input Data
  output:
    - name: out
      title: Output

入力ポートと出力ポートを備えた UDTF:

ports:
  input:
    - name: input_data
      title: Input Data
  output:
    - name: clustered_data
      title: Clustered Results

複数の入力とオプションのポートを持つ python-run-function:

ports:
  input:
    - name: main_data
      title: Main Data
    - name: reference_data
      title: Reference Table
      required: false
  output:
    - name: joined_output
      title: Joined Output

Config

config フィールドは JSON スキーマ オブジェクトです。 各構成フィールドは、スキーマ内のプロパティとして定義します。 この形式を使用すると、 enum、 minimum、 maximum、 examplesなどの標準的な JSON スキーマ検証機能にアクセスできます。

config オブジェクトには、type: objectとproperties マップが必要です。 必要に応じて、 required (必要なプロパティ名の配列) と additionalPropertiesを含めることができます。

config:
  type: object
  properties:
    cluster_count:
      type: number
      title: Number of Clusters
      description: How many clusters to create
      default: 3
      minimum: 1
      maximum: 100
    algorithm:
      type: string
      title: Algorithm
      description: Clustering algorithm to use
      enum: ['kmeans', 'dbscan', 'hierarchical']
      default: kmeans
    feature_col:
      type: string
      title: Feature Column
      description: Column to use as input
      format: expression
      x-ui:
        widget: expression
        port: data
  required: [cluster_count, feature_col]
  additionalProperties: false

Config プロパティ フィールド

config.properties オブジェクトの各プロパティは、次の標準 JSON スキーマ フィールドをサポートしています。

フィールド タイプ 説明
type 文字列 データ型: string、 number、 integer、 boolean、 array、または object。
title 文字列 人間が判読できるラベルが UI に表示されます。
description 文字列 ユーザーに表示されるヘルプ テキスト。
default いかなる／どれでも／任意の フィールドの既定値。
examples アレイ フィールドの値の例。
enum アレイ 許可される値の一覧を修正しました。
format 文字列 セマンティック型ヒント。 「 値の書式設定」を参照してください。
minimum number 最小値 ( number 型と integer 型の場合)。
maximum number 許容される最大値 ( number 型と integer 型の場合)。
items オブジェクト 配列要素のスキーマ ( type が arrayされている場合)。
properties オブジェクト 入れ子になったプロパティ定義 ( type が objectされている場合)。
required アレイ 必要な入れ子になったプロパティ名の一覧 ( type が objectされている場合)。

minLength、maxLength、pattern、constなどのその他の標準 JSON スキーマ フィールドもサポートされています。

値の書式設定

config プロパティの format フィールドには、値の解釈方法を Lakeflow Designer に指示するセマンティック型ヒントが用意されています。 これらのヒントにより、特殊な UI の動作と検証が可能になります。

フォーマット 説明
expression 列参照または SQL 式。
table_source テーブル ソース参照。
file_source ファイル ソース参照。
column_expressions 列式。
sort_expressions 式を並べ替える。
aggregation_expressions 集計式。
ai_function_expressions AI 関数式。
is_preview 自動プレビュー モード フラグ。 Lakeflow Designer は、ワークフロープレビュー中にこれを true に設定します。 config プロパティ名は任意です。 format: is_preview タグのみが重要です。 これを使用して、プレビュー中に外部 API 呼び出しなどの副作用をスキップします。
string[] 文字列配列。

UI ウィジェット

ウィジェットは、Lakeflow Designer インターフェイスでの構成フィールドのレンダリング方法をカスタマイズします。 各構成プロパティの x-ui プロパティでウィジェットを定義します。 ウィジェットを省略すると、Lakeflow Designer はデータ型に基づいて既定のウィジェットを使用します。

Widget データの種類 説明
input 文字列 1 行のテキスト入力。
textarea 文字列 複数行のテキスト領域。 省略可能な rows プロパティをサポートします。
checkbox boolean [標準] チェック ボックス。
toggle boolean 切り替えスイッチ。
number number/integer 省略可能な制約を持つ数値入力。
slider number/integer 数値範囲のビジュアル スライダー。 省略可能な step プロパティをサポートします。
select 文字列 単一選択ドロップダウン。 optionsSource が必要です。
multi-select アレイ 複数選択ドロップダウン。 optionsSource が必要です。
expression 文字列 列/式セレクター。 port が必要です。

input

1 行テキスト入力フィールド。

api_endpoint:
  type: string
  title: API Endpoint
  x-ui:
    widget: input

textarea

長いコンテンツ用の複数行テキスト領域。 高さを制御する省略可能な rows プロパティをサポートします。

message_body:
  type: string
  title: Message Body
  x-ui:
    widget: textarea
    rows: 4

checkbox

ブール値の標準チェック ボックス。

send_notification:
  type: boolean
  title: Send Notification
  default: false
  x-ui:
    widget: checkbox

toggle

ブール値の切り替えスイッチ。

enable_logging:
  type: boolean
  title: Enable Logging
  default: true
  x-ui:
    widget: toggle

number

数値入力フィールド。 範囲を制限するには、プロパティ自体に minimum と maximum を使用します。

num_clusters:
  type: number
  title: Number of Clusters
  default: 3
  minimum: 1
  maximum: 100
  x-ui:
    widget: number

slider

範囲内の数値を選択するためのビジュアル スライダー。 プロパティのminimumとmaximumを使用して範囲を設定し、stepでx-uiして増分を制御します。

confidence_threshold:
  type: number
  title: Confidence Threshold
  default: 0.8
  minimum: 0
  maximum: 1
  x-ui:
    widget: slider
    step: 0.05

select

単一選択ドロップダウン。 ドロップダウン値の取得場所を定義するには、 optionsSource が必要です。 「オプション ソース」を参照してください。

aggregation_type:
  type: string
  title: Aggregation Type
  x-ui:
    widget: select
    optionsSource:
      type: static
      values: ['sum', 'avg', 'min', 'max', 'count']

multi-select

複数の値を選択するための複数選択ドロップダウン。 プロパティのtype: arrayでitems: { type: string }を使用します。 optionsSourceが必要です。 「オプション ソース」を参照してください。

feature_columns:
  type: array
  title: Feature Columns
  items:
    type: string
  x-ui:
    widget: multi-select
    optionsSource:
      type: inputColumns
      port: input_data

expression

ユーザーが入力データから列を選択したり、カスタム SQL 式を書き込んだりできる列/式セレクター。 プロパティにformat: expressionを設定し、portで入力x-uiを指定します。 これは便利です。

  • ユーザーが入力データから列を選択する必要がある場合。
  • ユーザーがカスタム SQL 式を記述する場合。
  • パイプライン内の動的データを参照するパラメーターの場合。
amount:
  type: string
  title: Amount
  format: expression
  x-ui:
    widget: expression
    port: input_data

オプション ソース

selectウィジェットとmulti-selectウィジェットの場合は、ドロップダウン オプションがoptionsSourceを使用する場所を定義する必要があります。 static (YAML で定義された固定リスト) とinputColumns (入力ポートの列名) の 2 つのソースがあります。

静的オプション

YAML で定義されている値の固定リスト。

optionsSource:
  type: static
  values: ['option1', 'option2', 'option3']
プロパティ タイプ 必須 説明
type 文字列 Yes staticである必要があります。
values アレイ Yes ドロップダウンの文字列値の配列。

入力列

入力ポートの列名をドロップダウンに動的に設定します。

optionsSource:
  type: inputColumns
  port: input_data
プロパティ タイプ 必須 説明
type 文字列 Yes inputColumnsである必要があります。
port 文字列 Yes 列名を取得する入力ポートの名前。 定義されている入力ポートの 1 つの name と一致する必要があります。

run_function

run_function プロパティを使用すると、Pythonコードを python-run-function 演算子の YAML 構成に直接埋め込むことができます。 これにより、個別の Unity カタログ関数を登録する必要がなくなります。

run_function:
  type: inline
  code: |
    def run(config, inputs, spark):
        df = inputs["data"]
        threshold = config["threshold"]
        return {"out": df.filter(df["score"] > threshold)}
プロパティ タイプ 必須 説明
type 文字列 Yes inlineである必要があります。
code 文字列 Yes ソース コードPythonします。 run()関数を定義する必要があります。

run()関数は、次の 3 つの引数を受け取ります。

  • config: UI でユーザーによって設定された構成値のディクショナリ。
  • inputs: 入力ポート名を DataFrames にマッピングするディクショナリ。
  • spark: アクティブな SparkSession。

この関数は、出力ポート名を DataFrames にマッピングするディクショナリを返す必要があります。 キーは、nameで定義されている各出力ポートのports.output フィールドと正確に一致する必要があります。 たとえば、 outという名前の出力ポートを使用するとします。

return {"out": result_df}

複数の出力ポートを使用する場合:

return {"match": match_df, "rest": rest_df}

environment

environment プロパティは、python-run-function 演算子のPython環境を指定します。 これを使用して、環境バージョンをピン留めし、pip 依存関係を宣言します。

environment:
  environment_version: '4'
  dependencies:
    - 'scikit-learn>=1.3'
    - 'pandas>=2.0'
プロパティ タイプ 必須 説明
environment_version 文字列 いいえ 基本Pythonランタイムとプレインストール済みライブラリを設定するサーバーレス環境バージョン。 利用可能なバージョンについては、 環境バージョンを参照してください。 たとえば、「 "4" 」のように入力します。
dependencies 文字列の配列 いいえ pip 依存関係指定子の一覧。 各エントリは、標準の pip 構文 (たとえば、 "pandas>=2.0") に従います。

完全な例

UC ベースの UDF

この例では、複合関心を計算する Unity カタログベースの UDF 演算子を定義します。

schema: user-defined-operator-v0.1.0
type: uc-udf
name: Compound Interest
id: finance.compound_interest
version: '1.0.0'
description: >
  Calculates compound interest based on principal, rate, and time period.

config:
  type: object
  properties:
    principal:
      type: string
      title: Principal Amount
      format: expression
      x-ui:
        widget: expression
        port: input_data

    annual_rate:
      type: number
      title: Annual Interest Rate
      default: 5.0
      minimum: 0
      maximum: 100
      x-ui:
        widget: number

    years:
      type: number
      title: Number of Years
      default: 10
      minimum: 1
      maximum: 50
      x-ui:
        widget: slider
        step: 1

    compound_frequency:
      type: string
      title: Compounding Frequency
      default: 'monthly'
      x-ui:
        widget: select
        optionsSource:
          type: static
          values: ['daily', 'monthly', 'quarterly', 'annually']
  required: [principal, annual_rate]
  additionalProperties: false

ports:
  input:
    - name: input_data
      title: Input Data
  output:
    - name: out
      title: Output

Python run-function 演算子

この例では、K-Means クラスタリングを使用して顧客をセグメント化する python-run-function 演算子を定義します。

schema: user-defined-operator-v0.1.0
type: python-run-function
name: Customer Segmentation
id: ml.customer_segmentation
version: '1.2.0'
description: >
  Segments customers into groups based on selected features
  using K-Means clustering. Returns customer IDs with their
  assigned segment numbers.

config:
  type: object
  properties:
    num_segments:
      type: integer
      title: Number of Segments
      description: How many customer segments to create
      default: 3
      minimum: 2
      maximum: 20
      x-ui:
        widget: number
    customer_id_column:
      type: string
      title: Customer ID Column
      description: Column containing customer identifiers
      x-ui:
        widget: select
        optionsSource:
          type: inputColumns
          port: customer_data
    feature_columns:
      type: array
      title: Feature Columns
      description: Columns to use for segmentation
      items:
        type: string
      x-ui:
        widget: multi-select
        optionsSource:
          type: inputColumns
          port: customer_data
    normalize_features:
      type: boolean
      title: Normalize Features
      description: Whether to normalize feature values before clustering
      default: true
      x-ui:
        widget: toggle
  required: [num_segments, customer_id_column, feature_columns]
  additionalProperties: false

ports:
  input:
    - name: customer_data
      title: Customer Data
      mime: application/vnd.databricks.dataframe
  output:
    - name: segmented_customers
      title: Segmented Customers

run_function:
  type: inline
  code: |
    def run(config, inputs, spark):
        from pyspark.ml.feature import VectorAssembler, StandardScaler
        from pyspark.ml.clustering import KMeans

        df = inputs["customer_data"]
        id_col = config["customer_id_column"]
        features = config["feature_columns"]
        k = config["num_segments"]
        normalize = config.get("normalize_features", True)

        assembler = VectorAssembler(inputCols=features, outputCol="features_vec")
        assembled = assembler.transform(df)

        if normalize:
            scaler = StandardScaler(inputCol="features_vec", outputCol="scaled_features")
            model = scaler.fit(assembled)
            assembled = model.transform(assembled)
            feature_col = "scaled_features"
        else:
            feature_col = "features_vec"

        kmeans = KMeans(k=k, featuresCol=feature_col, predictionCol="segment")
        result = kmeans.fit(assembled).transform(assembled)

        return {"segmented_customers": result.select(id_col, "segment")}

environment:
  environment_version: '4'
  dependencies:
    - 'scikit-learn>=1.3'

クイック リファレンス

必須のルート プロパティ

  • schema: user-defined-operator-v0.1.0
  • name: 表示名
  • id: 一意識別子
  • description: 演算子の動作
  • config: JSON Schema オブジェクト
  • type: uc-udf、 uc-udtf、または python-run-function
  • version: 作成者が定義したバージョン文字列

省略可能なルート プロパティ

  • ports: 入力ポートと出力ポートの定義
  • run_function: インライン Python コード (python-run-function のみ)
  • environment: Python環境と依存関係 (python-run-function のみ)

プロパティのデータ型の構成

string | boolean | number | integer | array | object

UI ウィジェット

input | textarea | checkbox | toggle | number | slider | select | multi-select | expression

オプション ソース

static (固定値) | inputColumns (入力ポートから)

値の書式設定

expression | table_source | file_source | column_expressions | sort_expressions | aggregation_expressions | ai_function_expressions | is_preview | string[]

  • Last updated on