メインコンテンツまでスキップ
に最終更新

Lakeflow Designerユーザー定義演算子のロジックを記述する

オペレーターのロジック は、ユーザー定義演算子のランするコードです。python-run-function演算子の場合、インラインのPython run()関数です。uc-udfまたはuc-udtf演算子の場合、SQLまたはPythonで記述されたUnity Catalog関数です。

ロジックの記述は、 共有ユーザー定義オペレーターの構築のステップの1つです。ロジックを作成してテストした後、 YAML でオペレーターを記述してLakeflow Designer がUI でそれを表示する方法を理解できるようにし、 登録して オペレーター ペインに表示されるようにします。1つのオペレーターのすべてのステップをまとめて確認するには、 「チュートリアル: 複利計算オペレーター」などのチュートリアルに従ってください。

python-ラン-function演算子を記述します​

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

Python
def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config : UI からユーザーが構成した値（プロパティ名でキー指定）。
  • inputs : 入力DataFrames。入力ポート name をキーとします。
  • spark : アクティブな SparkSession。
  • 戻り値 ：出力ポート name の値から DataFrames へのマッピングを行うディクショナリ。

次の例では、入力 DataFrame から行をフィルタリングします:

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

登録する前に関数をテストするには、サンプルデータを使用してノートブックで関数を呼び出します。

Python
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 に追加します。

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関数：

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でラップされたもの）：

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つの値を処理し、計算された値を返します。UDTFは、テーブルを1行ずつ処理し、すべての行にわたって状態を維持できます。列レベルの変換にはuc-udfを使用し、機械学習クラスター化や集計などの操作にはuc-udtfを使用します。

UDTF ハンドラークラスは、各入力行で実行される eval() を定義します。オプションで、状態をセットアップする __init__() と、すべての行の後に最終結果を生成する terminate() も定義します。Unity Catalog の Python ユーザー定義テーブル関数 (UDTF)を参照してください。

Python
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.
        ...
注記

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

プレビュー中の副作用をスキップする​

Lakeflow Designer は、デザイン モードでのプレビューをサポートします。外部APIsを呼び出すか外部システムに書き込むオペレーターの場合、is_preview 設定プロパティを追加して、プレビュー中に副作用を回避できるようにします。ユーザーは ラン をクリックして、副作用とともにオペレーターを実行します。組み込みPython演算子は config["is_preview"] を自動的に読み取るため、このプロパティは不要です。Pythonプレビューモードを参照してください。

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

Lakeflow Designerは、プレビュー中にこの値を自動的に true に設定します。副作用を回避するためにロジックで確認します：

Python
# In a python-run-function
if config.get("is_preview"):
    return {"out": inputs["in"]}
SQL
-- 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接続を使用して認証情報を安全に格納します。

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

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

WorkspaceClient でワークスペースリソースにアクセスする​

python-run-function オペレーターの場合、Databricks WorkspaceClient を使用してワークスペースリソースおよび外部 APIs にアクセスできます。

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

追加リソース​

このページの見出し