Справочник по определяемой пользователем оператору YAML

Определяемые пользователем операторы в Конструкторе Lakeflow определяются в YAML. Все типы операторов (uc-udf, uc-udtfи python-run-function) используют схему user-defined-operator-v0.1.0 , которая определяет поля конфигурации с помощью формата схемы JSON.

Сведения о создании определяемых пользователем операторов см. в разделе "Определяемые пользователем операторы" в конструкторе Lakeflow.

Корневые свойства

Каждый файл 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'

Порты

Порты определяют, как оператор подключается к другим операторам в конвейере. Объект 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

Принимаются только задокументированные свойства порта. Неизвестные ключи (например, устаревшее 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. Каждое поле конфигурации определяется как свойство в схеме. Этот формат предоставляет доступ к стандартным функциям проверки схемы JSON, таким как enum, и minimummaximumexamples.

Объект 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.properties поля схемы JSON:

Другие стандартные поля схемы JSON, такие как minLength, maxLengthpatternи const также поддерживаются.

Форматирование значений

Поле format свойства конфигурации предоставляет указание семантического типа, указывающее конструктор Lakeflow, как интерпретировать значение. Эти указания позволяют использовать специализированное поведение пользовательского интерфейса и проверку.

Мини-приложения пользовательского интерфейса

Мини-приложения настраивают отображение поля конфигурации в интерфейсе конструктора Lakeflow. Определите мини-приложения в свойстве x-ui для каждого свойства конфигурации. Если вы опустите мини-приложение, Конструктор Lakeflow использует мини-приложение по умолчанию на основе типа данных.

input

Однострочного текстового поля ввода.

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 для свойства, чтобы задать диапазон, а stepx-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 для свойства и укажите входные данныеportx-ui. Это полезно:

• Когда пользователь должен выбрать столбец из входных данных.
• Если пользователю может потребоваться написать пользовательское выражение SQL.
• Для параметров, ссылающихся на динамические данные в конвейере.

amount:
  type: string
  title: Amount
  format: expression
  x-ui:
    widget: expression
    port: input_data

Источники параметров

Для select и multi-select мини-приложений необходимо определить, откуда optionsSourceприходят параметры раскрывающегося списка. Существует два источника: static (фиксированный список, определенный в YAML) и inputColumns (имена столбцов из входного порта).

Статические параметры

Фиксированный список значений, определенных в YAML.

optionsSource:
  type: static
  values: ['option1', 'option2', 'option3']

Входные столбцы

Динамически заполняет раскрывающийся список именами столбцов из входного порта.

optionsSource:
  type: inputColumns
  port: input_data

run_function

Свойство run_function позволяет внедрять код Python непосредственно в конфигурацию YAML для операторов python-run-function. Это устраняет необходимость зарегистрировать отдельную функцию каталога Unity.

run_function:
  type: inline
  code: |
    def run(config, inputs, spark):
        df = inputs["data"]
        threshold = config["threshold"]
        return {"out": df.filter(df["score"] > threshold)}

Функция run() получает три аргумента:

• config: словарь значений конфигурации, заданных пользователем в пользовательском интерфейсе.
• inputs: словарь сопоставляет имена входных портов с Кадрами данных.
• spark: активная sparkSession.

Функция должна возвращать имена выходных портов словаря в кадры данных. Ключи должны точно соответствовать name полю каждого выходного порта, определенного в ports.output. Например, с портом вывода с именем out:

return {"out": result_df}

С несколькими выходными портами:

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

environment

Свойство environment указывает среду Python для операторов python-run-function. Используйте его для закрепления версии среды и объявления зависимостей pip.

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

Полные примеры

UDF на основе UC

В этом примере определяется оператор UDF на основе каталога Unity, который вычисляет составные интересы.

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

оператор run-function Python

В этом примере определяется python-run-function оператор, который сегментирует клиентов с помощью кластеризации K-Средних.

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

Мини-приложения пользовательского интерфейса

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[]