Lakeflow Designer kullanıcı tanımlı operatörleri için mantık yazın

Operatör mantığı, kullanıcı tanımlı bir operatör çalıştırıldığında çalışan koddur. Bir python-run-function operatör için bu, satır içi bir Python run() fonksiyonudur. Bir uc-udf veya uc-udtf operatörü için bu, SQL veya Python ile yazılmış bir Unity Kataloğu işlevidir.

Mantığı yazmak, paylaşılan kullanıcı tanımlı bir operatör oluşturma sürecindeki adımlardan biridir. Mantığı yazıp test ettikten sonra, operatörü YAML'de tanımlarsınız ki Lakeflow Designer arayüzünde nasıl göstereceğini bilsin, sonra kaydedersiniz ki Operatörler panelinde görünsün. Tek bir operatör için tüm adımları birlikte görmek için, Eğitim: Bileşik faiz hesaplayıcı operatörü gibi bir eğitimi izleyin.

Bir python-run-function operatörü yaz

Her python-run-function işleç bir run() işlev tanımlamalıdır:

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: Kullanıcı arabiriminden kullanıcı tarafından yapılandırılan ve özellik adına göre anahtarlanan değerler.
  • inputs: Giriş bağlantı noktası nametarafından anahtarlanan Giriş Veri Çerçeveleri.
  • spark: Aktif SparkSession.
  • Döndürür: Çıkış bağlantı noktası name değerlerini DataFrame'lerle eşleyen bir sözlük.

Aşağıdaki örnek, bir giriş DataFrame'indeki satırları filtreler:

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

Fonksiyonu kaydetmeden önce test etmek için, örnek verilerle bir notebook'ta çağırın:

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()

Operatörünüz harici pip paketleri gerektiriyorsa alanı YAML'ye ekleyin environment :

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

Bir UDF veya UDTF operatörü yaz

Unity Kataloğu fonksiyonlarını SQL veya Python'da yazabilirsiniz. Python işlevleri sql CREATE FUNCTION deyiminde sarmalanmıştır:

SQL işlevi:

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 fonksiyonu (SQL ile sarmalanmış):

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'ler tek seferde tek bir değeri işler ve hesaplanan bir değer döndürür. UDTF'ler tabloları satır satır işler ve tüm satırlar boyunca durumu koruyabilir. Sütun düzeyindeki dönüşümler için uc-udf ve ML kümeleme veya toplulaştırma gibi işlemler için uc-udtf kullanın.

Bir UDTF handler sınıfı, her giriş satırı için çalışan eval() öğesini tanımlar. İsteğe bağlı olarak, durumu ayarlamak için terminate() ve tüm satırlardan sonra nihai sonuçları vermek için __init__() de tanımlar. Bkz. Unity Catalog'da Python kullanıcı tanımlı tablo işlevleri (UDTF'ler).

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.
        ...

Uyarı

UDTF dönüş tabloları sabit ve açık türlere sahip olmalıdır. Dönüş yapılandırmasında giriş sütunu türlerine başvuramazsınız.

Ön izlemelerde yan etkileri atlayın

Lakeflow Designer, tasarım modundayken önizlemeleri destekler. Dış API'leri çağıran veya dış sistemlere yazan işleçler için, önizleme sırasında yan etkileri atlayabileceğiniz bir is_preview yapılandırma özelliği ekleyin. Kullanıcılar, operatörü yan etkileriyle çalıştırmak için Çalıştır tuşuna tıklar. Dahili Python operatörü bu özelliğe ihtiyaç duymaz, çünkü config["is_preview"] öğesini otomatik olarak okur. Python önizleme moduna bakınız.

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

Lakeflow Designer, önizleme sırasında bu değeri otomatik olarak olarak true olarak ayarlar. Yan etkileri atlamak için bunu mantığınızda kontrol edin:

# 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 bağlantılarıyla harici API'leri çağırın

Harici API'leri çağıran SQL Unity Kataloğu işlevlerine dayanan kullanıcı tanımlı operatörler için, kimlik bilgilerini güvenli bir şekilde depolamak üzere Unity Catalog HTTP bağlantılarını kullanın:

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

Ardından, SQL UDF'nizde http_request() işleviyle bağlantıyı kullanın. Ayrıntılar için bkz. Dış HTTP hizmetlerine bağlanma.

WorkspaceClient ile çalışma alanı kaynaklarına erişim

python-run-function işleçleri için çalışma alanı kaynaklarına ve dış API'lere erişmek için Azure Databricks WorkspaceClient kullanabilirsiniz:

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

Ek kaynaklar

  • Son güncelleştirme: