Lakeflow Designerのカスタムオペレーター用ロジックを書く

オペレーター ロジックとは、カスタム オペレーターが実行するコードのことです。 python-run-functionオペレーターの場合はインラインのPython run()関数です。 uc-udf または uc-udtf オペレーターの場合、これは SQL または Python で記述された Unity Catalog の関数です。

ロジックを書くことは共有カスタム演算子を構築する工程の一つです。 ロジックを書いてテストした後、YAMLでオペレーターを記述し、Lakeflow DesignerがUIで表示方法を認識できるようにしてから、登録し、Operatorsペインに表示されるようにします。 1 つのオペレーターに関するすべての手順を一覧で確認するには、チュートリアル: 複利計算機オペレーターなどのチュートリアルを参照してください。

Python の実行関数オペレーターを作成する

すべての python-run-function 演算子は、 run() 関数を定義する必要があります。

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: UI からユーザーが構成した値。プロパティ名でキー指定されます。
  • inputs: 入力ポート nameによってキー指定された入力データフレーム。
  • spark: アクティブな SparkSession。
  • 戻り値: 出力ポート name 値を DataFrames にマッピングするディクショナリ。

次の例では、入力 DataFrame から行をフィルター処理します。

def run(config, inputs, spark):
    df = inputs["in"]
    filtered = df.filter(config["filter_expression"])
    return {"out": filtered}

関数を登録する前にテストするには、サンプルデータ付きのノートブックで呼び出してください:

test_df = spark.createDataFrame(
    [("Alice", 100), ("Bob", 200)],
    ["name", "amount"]
)

result = run(
    config={"filter_expression": "amount > 100"},
    inputs={"in": test_df},
    spark=spark
)

result["out"].show()

オペレーターが外部 pip パッケージを必要とする場合は、 environment フィールドを YAML に追加します。

environment:
  environment_version: '4'
  dependencies:
    - requests==2.31.0
    - beautifulsoup4==4.12.0

UDFまたはUDTFオペレーターを作成してください

Unity Catalogの関数はSQLやPythonで書けます。 Python関数は、SQL CREATE FUNCTION ステートメントでラップされます。

SQL 関数:

CREATE OR REPLACE FUNCTION my_catalog.my_schema.calculate_bmi(weight_kg DOUBLE, height_m DOUBLE)
RETURNS DOUBLE
LANGUAGE SQL
RETURN
  SELECT weight_kg / (height_m * height_m);

Python 関数（SQL でラップされた）:

CREATE OR REPLACE FUNCTION my_catalog.my_schema.calculate_bmi(weight_kg DOUBLE, height_m DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
AS $$
  return weight_kg / (height_m ** 2)
$$;

UDF は一度に 1 つの値を処理し、計算値を返します。 UDF は行ごとにテーブルを処理し、すべての行で状態を維持できます。 列レベルの変換には uc-udf を使用し、ML クラスタリングや集計などの操作には uc-udtf を使用します。

UDTFハンドラークラスは、各入力行ごとに実行される eval() を定義します。 オプションとして、状態を設定する __init__() と、すべての行の処理後に最終結果を出す terminate() も定義しています。 Unity カタログの Python ユーザー定義テーブル関数 (UDF) を参照してください。

class MyOperator:
    def __init__(self):
        # Optional. Called before processing - initialize any values needed.
        ...

    def eval(self, row):
        # Required. Called one time per input row - process or accumulate data here.
        ...

    def terminate(self):
        # Optional. Called after all rows - perform final calculations and yield results.
        ...

Note

UDTF 戻り値テーブルには、固定の明示的な型が必要です。 戻り値の構成設定で入力列の型を参照することはできません。

プレビュー中の副作用はスキップしましょう

Lakeflow Designer では、デザイン モード中のプレビューがサポートされます。 外部 API を呼び出したり、外部システムに書き込んだりする演算子の場合は、プレビュー中に副作用をスキップできるように、 is_preview 構成プロパティを追加します。 ユーザーは実行をクリックして、そのオペレーターを副作用込みで実行します。 内蔵のPython演算子はこのプロパティを必要としません。なぜなら自動的にconfig["is_preview"]を読み込むからです。 Pythonプレビューモードをご覧ください。

config:
  type: object
  properties:
    is_preview:
      type: boolean
      format: is_preview
      default: false

Lakeflow Designer では、プレビュー中にこの値が自動的に true に設定されます。 副作用をスキップするために、ロジック内でチェックします:

# In a python-run-function
if config.get("is_preview"):
    return {"out": inputs["in"]}

-- In a UC function (SQL)
CASE WHEN is_preview THEN 'preview' ELSE /* actual work */ END

Unity CatalogのHTTP接続で外部APIを呼び出す

外部APIを呼び出すSQL Unity Catalog関数ベースのカスタムオペレーターには、認証情報を安全に保存するために、Unity CatalogのHTTP接続を使用してください:

CREATE CONNECTION my_api_connection TYPE HTTP OPTIONS (
  host 'https://api.example.com',
  port '443',
  base_path '/v1/',
  bearer_token 'your-token-here'
);

次に、 http_request() 関数を使用して、SQL UDF 内の接続を使用します。 詳細については、「 外部 HTTP サービスへの接続」を参照してください。

ワークスペースリソースにはWorkspaceClientを使ってアクセスできます

python-run-function 演算子の場合は、Azure Databricks WorkspaceClient を使用して、ワークスペース リソースと外部 API にアクセスできます。

def run(config, inputs, spark):
    from databricks.sdk import WorkspaceClient
    w = WorkspaceClient()
    # Use w to access workspace resources

追加リソース

  • 最終更新: