Operator yang ditentukan pengguna di Lakeflow Designer

Lakeflow Designer memungkinkan Anda membuat operator yang ditentukan pengguna yang muncul langsung di kanvas bersama operator bawaan. Gunakan untuk memperluas Lakeflow Designer dengan logika bisnis, perhitungan, atau integrasi Anda sendiri.

Ada tiga jenis operator yang ditentukan pengguna:

  • python-run-function: Sebuah file YAML terpisah dengan kode Python inline yang disimpan di ruang kerja. Terbaik untuk transformasi tingkat DataFrame dan integrasi eksternal. Izin akses dikelola pada tingkat file di ruang kerja.
  • uc-udf: Membungkus fungsi skalar Unity Catalog. Terbaik untuk transformasi tingkat kolom. Akses diatur oleh izin Katalog Unity.
  • uc-udtf: Membungkus fungsi bernilai tabel Unity Catalog. Terbaik untuk transformasi tingkat tabel seperti pengklusteran dan agregasi ML. Akses diatur oleh izin Katalog Unity.
Feature python-run-function uc-udf uc-udtf
Contoh kasus penggunaan Transformasi DataFrame, integrasi API, pemberitahuan email Perhitungan tingkat kolom (BMI, suku bunga) Pengklusteran ML, agregasi di seluruh baris
Input DataFrame Nilai tunggal Seluruh tabel, baris demi baris
Hasil DataFrame Nilai tunggal Tabel (beberapa baris)
Memerlukan fungsi Katalog Unity No Yes Yes
Pengelolaan akses Izin akses file ruang kerja Izin akses Unity Catalog (EXECUTE, USE SCHEMA) Izin akses Unity Catalog (EXECUTE, USE SCHEMA)
Bahasa yang didukung Hanya Python SQL atau Python dalam pembungkus SQL SQL atau Python dalam pembungkus SQL

Cara kerja operator yang ditentukan pengguna

Operator yang ditentukan pengguna terdiri dari:

  • Logika operator: Kode yang berjalan saat operator dijalankan. Ini bisa menjadi fungsi Python run() sebaris (untuk python-run-function) atau fungsi Katalog Unity (untuk uc-udf dan uc-udtf).
  • Konfigurasi YAML: Memberi tahu Lakeflow Designer cara menyajikan operator di UI, termasuk nama, deskripsi, parameter input, widget UI, dan port operator. Semua jenis operator menggunakan user-defined-operator-v0.1.0 skema.
  • File pendaftaran: Entri di .user_defined_operators.yaml yang memungkinkan Lakeflow Designer menemukan operator.

Logika operator

Python menjalankan logika operator yang ditentukan pengguna fungsi

Setiap python-run-function operator harus menentukan run() fungsi:

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: Nilai yang dikonfigurasi pengguna melalui UI, diindeks berdasarkan nama properti.
  • inputs: DataFrame input, diidentifikasi berdasarkan port input name.
  • spark: SparkSession aktif.
  • Hasil pengembalian: Kamus yang memetakan nilai name port output ke DataFrame.

Contoh berikut memfilter baris dari DataFrame input:

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

Jika operator Anda memerlukan paket pip eksternal, tambahkan environment bidang ke YAML:

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

Logika operator UDF dan UDTF

Anda dapat menulis fungsi UC di SQL atau Python. fungsi Python dibungkus dalam pernyataan SQL CREATE FUNCTION:

Fungsi 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);

fungsi Python (dibungkus dalam 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 memproses satu nilai sekaligus dan mengembalikan nilai terhitung. UDTF memproses tabel baris demi baris dan dapat mempertahankan status di semua baris. Gunakan uc-udf untuk transformasi tingkat kolom dan uc-udtf untuk operasi seperti pengklusteran atau agregasi ML.

Selain itu, UDTF mengharuskan Anda menentukan tiga metode utama: __init__(), , eval()dan terminate():

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

    def eval(self, row, id_column, columns, k):
        # Called one time per input row - accumulate data here.

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

Note

Tabel pengembalian UDTF harus memiliki jenis eksplisit tetap. Anda tidak dapat mereferensikan jenis kolom input dalam konfigurasi pengembalian.

Konfigurasi YAML

Konfigurasi YAML memberi tahu Lakeflow Designer cara menyajikan operator di UI. Ini mendefinisikan nama operator, deskripsi, parameter input, widget UI, dan port. Setiap bidang konfigurasi adalah properti dengan tipe, judul, dan petunjuk widget opsional x-ui:

config:
  type: object
  properties:
    my_param:
      type: string
      title: My Parameter
      x-ui:
        widget: input
    my_expression:
      type: string
      title: Column
      format: expression
      x-ui:
        widget: expression
        port: in
    my_number:
      type: number
      title: Count
      default: 10
      minimum: 0
      maximum: 100
  required:
    - my_param
    - my_expression

Untuk detail lengkap tentang skema YAML, termasuk semua jenis widget dan opsi konfigurasi, lihat Referensi YAML operator yang ditentukan pengguna.

Pelabuhan

Port mendefinisikan masukan dan keluaran untuk operator Anda:

ports:
  input:
    - name: in
      title: Input Data
      mime: application/vnd.databricks.dataframe
      required: true
      allowMultiple: false
  output:
    - name: out
      title: Output Data

YAML untuk Python menjalankan operator fungsi

Untuk operator python-run-function, file YAML berdiri sendiri dan menyertakan field run_function dengan kode Python inline:

schema: user-defined-operator-v0.1.0
type: python-run-function
name: Filter Rows
id: filter_rows
version: '1.0.0'
description: Filters rows based on a SQL expression.
config:
  type: object
  properties:
    filter_expression:
      type: string
      title: Filter Expression
      x-ui:
        widget: input
  required:
    - filter_expression
ports:
  input:
    - name: in
      title: Input
  output:
    - name: out
      title: Output
run_function:
  type: inline
  code: |
    def run(config, inputs, spark):
        df = inputs["in"]
        filtered = df.filter(config["filter_expression"])
        return {"out": filtered}

YAML untuk fungsi Katalog Unity

Untuk operator berbasis UC, sematkan konfigurasi YAML sebagai komentar atau docstring dalam fungsi Anda.

Di SQL (gunakan /* ... */ komentar):

RETURN(/*
  schema: user-defined-operator-v0.1.0
  type: uc-udf
  name: Calculate BMI
  id: calculate_bmi
  version: "1.0.0"
  description: Calculates BMI from weight and height.
  config:
    type: object
    properties:
      weight_kg:
        type: string
        title: Weight (in kg)
        format: expression
        x-ui:
          widget: expression
          port: in
      height_m:
        type: string
        title: Height (in meters)
        format: expression
        x-ui:
          widget: expression
          port: in
    required:
      - weight_kg
      - height_m
  ports:
    input:
      - name: in
        title: Input Data
    output:
      - name: out
        title: Output
    */
  SELECT weight_kg / (height_m * height_m)
);

In Python (gunakan """ ... """ docstring):

AS $$
  """
  schema: user-defined-operator-v0.1.0
  type: uc-udf
  name: Calculate BMI
  id: calculate_bmi
  version: "1.0.0"
  description: Calculates BMI from weight and height.
  config:
    type: object
    properties:
      weight_kg:
        type: string
        title: Weight (in kg)
        format: expression
        x-ui:
          widget: expression
          port: in
      height_m:
        type: string
        title: Height (in meters)
        format: expression
        x-ui:
          widget: expression
          port: in
    required:
      - weight_kg
      - height_m
  ports:
    input:
      - name: in
        title: Input Data
    output:
      - name: out
        title: Output
  """

  return weight_kg / (height_m ** 2)
$$;

Daftarkan dan terapkan operator Anda ke Lakeflow Designer

Agar operator Anda muncul di Lakeflow Designer, daftarkan dalam .user_defined_operators.yaml file:

  • Tingkat ruang kerja: Tempatkan file di akar ruang kerja Anda untuk membuat operator terlihat oleh semua pengguna.
  • Tingkat pengguna: Tempatkan file di folder beranda pengguna Anda (/Workspace/Users/<user-name>/.user_defined_operators.yaml) untuk membuat operator hanya terlihat oleh Anda.

operators:bagian mendukung jalur file, referensi fungsi Unity Catalog, dan pola glob. Anda dapat mencampur jenis entri:

operators:
  # File path (python-run-function operators)
  - /Workspace/Users/me/udos/my_operator.yaml
  # Glob pattern (registers all matching files)
  - /Workspace/Users/me/udos/transforms/*.yaml
  # UC function reference (uc-udf and uc-udtf operators)
  - catalog: my_catalog
    schema: my_schema
    functionName: my_function

Memperbarui atau menghapus operator

Saat Anda mengubah kode operator, muat ulang operator yang ditentukan pengguna agar perubahan tersebut dimuat. Di tab Operator menu, klik ikon Refresh..

  • Jika operator tetap menggunakan version yang sama, memuat ulang akan memuat kode yang diperbarui.
  • Jika operator memiliki version baru, operator pada kanvas akan meminta Anda untuk memutakhirkan ke versi tersebut (atau tetap menggunakan versi saat ini) setelah Anda memuat ulang.

Untuk menghapus operator dari Lakeflow Designer, hapus entrinya dari .user_defined_operators.yaml. Untuk uc-udf operator dan uc-udtf , Anda juga dapat menghilangkan fungsi Unity Catalog yang mendasar dengan DROP FUNCTION jika Anda tidak lagi membutuhkannya.

Konfigurasi tingkat lanjut

Mode Pratinjau

Lakeflow Designer mendukung pratinjau saat dalam mode desain. Untuk operator yang memanggil API eksternal atau menulis data ke sistem eksternal, tambahkan properti konfigurasi is_preview agar Anda dapat mengabaikan efek samping selama pratinjau. Saat mode pratinjau diaktifkan, pengguna perlu secara eksplisit mengklik Jalankan untuk menjalankan operator dengan efek samping.

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

Lakeflow Designer secara otomatis mengatur nilai ini ke true selama pratinjau. Periksa dalam logika Anda untuk melewati efek samping:

# 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

Koneksi Katalog Unity

Untuk operator SQL berbasis UC yang memanggil API eksternal, gunakan koneksi HTTP Katalog Unity untuk menyimpan kredensial dengan aman:

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

Kemudian gunakan koneksi di SQL UDF Anda dengan http_request() fungsi . Untuk detailnya, lihat Menyambungkan ke layanan HTTP eksternal.

WorkspaceClient

Untuk operator python-run-function, Anda dapat menggunakan Azure Databricks WorkspaceClient untuk mengakses sumber daya ruang kerja dan API eksternal:

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

Buat operator python-run-function yang ditentukan pengguna secara lengkap

Langkah-langkah berikut memandu Anda membuat operator python-run-function dari awal.

Langkah 1: Tentukan logika

Tulis fungsi Anda run() di buku catatan:

from typing import Dict, Any

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
    from pyspark.sql import functions as F
    df = inputs["in"]
    result = df.withColumn(config["column_name"], F.current_timestamp())
    return {"out": result}

Langkah 2: Uji fungsi

Uji fungsi secara interaktif dengan data sampel:

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

result = run(
    config={"column_name": "processed_at"},
    inputs={"in": test_df},
    spark=spark
)

result["out"].show()

Langkah 3: Membuat konfigurasi YAML

Tentukan metadata operator, bidang konfigurasi, dan port dalam file YAML:

schema: user-defined-operator-v0.1.0
type: python-run-function
name: Add Timestamp
id: transforms.add_timestamp
version: '1.0.0'
description: Adds a timestamp column to the input DataFrame.
config:
  type: object
  properties:
    column_name:
      type: string
      title: Column Name
      default: processed_at
      x-ui:
        widget: input
  required:
    - column_name

Langkah 4: Gabungkan logika dan YAML

Tambahkan bidang run_function dan ports untuk membuat file YAML lengkap. Simpan ke ruang kerja Anda, misalnya /Workspace/Users/<user-name>/udos/add_timestamp.yaml:

schema: user-defined-operator-v0.1.0
type: python-run-function
name: Add Timestamp
id: transforms.add_timestamp
version: '1.0.0'
description: Adds a timestamp column to the input DataFrame.
config:
  type: object
  properties:
    column_name:
      type: string
      title: Column Name
      default: processed_at
      x-ui:
        widget: input
  required:
    - column_name
ports:
  input:
    - name: in
      title: Input
  output:
    - name: out
      title: Output
run_function:
  type: inline
  code: |
    from typing import Dict, Any

    def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
        from pyspark.sql import functions as F
        df = inputs["in"]
        result = df.withColumn(config["column_name"], F.current_timestamp())
        return {"out": result}

Langkah 5: Daftarkan operator

Tambahkan jalur file ke file Anda .user_defined_operators.yaml :

operators:
  - /Workspace/Users/<user-name>/udos/add_timestamp.yaml

Langkah 6: Gunakan operator di Lakeflow Designer

Buka Lakeflow Designer dan pastikan bahwa operator muncul di palet operator. Seret ke kanvas, sambungkan input, konfigurasikan nama kolom, dan jalankan pratinjau.

Buat operator yang ditentukan pengguna UC secara lengkap

Langkah-langkah berikut menjelaskan cara membuat operator uc-udf berbasis UC.

Langkah 1: Tentukan logika

Tulis dan uji logika fungsi Anda di buku catatan:

def double_value(input_value: float) -> float:
    if input_value is None:
        return None
    return input_value * 2

Langkah 2: Membuat konfigurasi YAML

Tentukan metadata operator, bidang konfigurasi, dan port:

schema: user-defined-operator-v0.1.0
type: uc-udf
name: Double Value
id: math.double_value
version: '1.0.0'
description: Doubles the input value
config:
  type: object
  properties:
    input_value:
      type: string
      title: Input Value
      format: expression
      x-ui:
        widget: expression
        port: input_data
  required:
    - input_value
ports:
  input:
    - name: input_data
      title: Input
  output:
    - name: out
      title: Output

Langkah 3: Gabungkan logika dan YAML

Buat fungsi Unity Catalog dengan YAML yang disematkan sebagai docstring:

CREATE OR REPLACE FUNCTION main.my_schema.double_value(input_value DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
AS $$
  """
  schema: user-defined-operator-v0.1.0
  type: uc-udf
  name: Double Value
  id: math.double_value
  version: "1.0.0"
  description: Doubles the input value
  config:
    type: object
    properties:
      input_value:
        type: string
        title: Input Value
        format: expression
        x-ui:
          widget: expression
          port: input_data
    required:
      - input_value
  ports:
    input:
      - name: input_data
        title: Input
    output:
      - name: out
        title: Output
  """

  def double_value(input_value: float) -> float:
      if input_value is None:
          return None
      return input_value * 2

  return double_value(input_value)
$$

Langkah 4: Uji fungsi

SELECT main.my_schema.double_value(5) AS result;
-- Should return: 10

Langkah 5: Daftarkan operator

Tambahkan referensi fungsi Katalog Unity ke file Anda .user_defined_operators.yaml :

operators:
  - catalog: main
    schema: my_schema
    functionName: double_value

Langkah 6: Gunakan operator di Lakeflow Designer

Buka Lakeflow Designer dan pastikan bahwa operator muncul di palet operator. Seret ke kanvas, sambungkan input, dan jalankan pratinjau.

Troubleshooting

Issue Solusi
Operator tidak muncul di Lakeflow Designer. Periksa apakah .user_defined_operators.yaml ada dan cantumkan fungsi atau jalur file Anda. Untuk operator python-run-function, verifikasi jalur file dan pastikan file YAML dapat diakses.
Validasi skema gagal. Verifikasi YAML Anda terhadap skema resmi di https://your-workspace.cloud.databricks.com/static/schemas/user-defined-operator-v0.1.0.json.
Izin ditolak. Untuk operator berbasis UC, verifikasi bahwa pengguna memiliki EXECUTE pada fungsi dan USE SCHEMA pada skema. Untuk operator python-run-function, pastikan pengguna memiliki izin baca ke file YAML.
python-run-function operator gagal pada runtime. Periksa apakah signature fungsi run() cocok dengan def run(config, inputs, spark). Verifikasi bahwa nama port dalam kode cocok dengan YAML dan bahwa kunci kamus pengembalian cocok dengan nilai port name output.
UDTF mengembalikan tipe yang salah. Jenis pengembalian UDTF harus eksplisit; Anda tidak dapat mereferensikan jenis kolom input.

Permissions

Persetujuan Kegunaan
Akses baca ke .user_defined_operators.yaml. Temukan operator.
Akses baca untuk file YAML (hanya python-run-function). Muat definisi operator.
EXECUTE pada fungsi Unity Catalog (hanya operator berbasis UC). Jalankan operator.
USE SCHEMA di skema (khusus operator berbasis UC). Akses skema tempat fungsi dibuat.
Izin lain Bergantung pada operator Anda, pengguna mungkin memerlukan izin lain. Misalnya, USE CONNECTION pada koneksi Unity Catalog untuk panggilan API HTTP.

Sumber daya tambahan

Jelajahi tutorial berikut:

Contoh Type Description
Pengirim email Gmail python-run-function Kirim data DataFrame sebagai lampiran email CSV melalui Gmail.
Kalkulator bunga majemuk uc-udf Hitung nilai investasi di masa mendatang menggunakan rumus bunga campuran.
Klasterisasi K-means uc-udtf Segmentasikan data ke dalam kluster menggunakan scikit-learn.
Kirim pesan Slack uc-udf Kirim pemberitahuan ke saluran Slack melalui API.
Semua widget UI uc-udf Operator referensi menampilkan semua widget UI yang tersedia.

Untuk referensi lengkap ke skema YAML, lihat Referensi YAML operator yang ditentukan pengguna.

  • Last updated on