Référence YAML de l’opérateur défini par l’utilisateur

Les opérateurs définis par l’utilisateur dans Lakeflow Designer sont définis dans YAML. Tous les types d’opérateurs (uc-udf, uc-udtfet python-run-function) utilisent le user-defined-operator-v0.1.0 schéma, qui définit les champs de configuration à l’aide du format de schéma JSON.

Pour plus d’informations sur la création d’opérateurs définis par l’utilisateur, consultez Les opérateurs définis par l’utilisateur dans Lakeflow Designer.

Propriétés racines

Chaque fichier YAML d’opérateur commence par un ensemble de propriétés racine qui identifient l’opérateur et définissent son comportement. L’exemple suivant montre la structure générale :

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

Les ports définissent la façon dont votre opérateur se connecte à d’autres opérateurs dans le pipeline. L’objet ports contient input et output tableaux.

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

Seules les propriétés de port documentées sont acceptées. Les clés inconnues (telles que le champ hérité label ) sont rejetées par la validation du schéma.

Exemples de ports

UDF avec ports d’entrée et de sortie :

ports:
  input:
    - name: in
      title: Input Data
  output:
    - name: out
      title: Output

UDTF avec des ports d’entrée et de sortie :

ports:
  input:
    - name: input_data
      title: Input Data
  output:
    - name: clustered_data
      title: Clustered Results

python-run-function avec plusieurs entrées et un port facultatif :

ports:
  input:
    - name: main_data
      title: Main Data
    - name: reference_data
      title: Reference Table
      required: false
  output:
    - name: joined_output
      title: Joined Output

Config

Le config champ est un objet json Schema. Vous définissez chaque champ de configuration en tant que propriété dans le schéma. Ce format vous donne accès aux fonctionnalités de validation de schéma JSON standard telles que enum, , minimummaximumet examples.

L’objet config doit avoir type: object et une properties carte. Vous pouvez éventuellement inclure required (tableau de noms de propriétés requis) et 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

Champs de propriété config

Chaque propriété de l’objet config.properties prend en charge les champs de schéma JSON standard suivants :

D’autres champs de schéma JSON standard tels que minLength, maxLength, patternet const sont également pris en charge.

Mettre en forme les valeurs

Le format champ d’une propriété de configuration fournit un indicateur de type sémantique qui indique au Concepteur Lakeflow comment interpréter la valeur. Ces indicateurs permettent un comportement et une validation spécialisés de l’interface utilisateur.

Widgets d’interface utilisateur

Les widgets personnalisent le rendu d’un champ de configuration dans l’interface du Concepteur Lakeflow. Définissez des widgets dans la x-ui propriété sur chaque propriété de configuration. Si vous omettez le widget, Lakeflow Designer utilise un widget par défaut basé sur le type de données.

input

Champ d’entrée de texte à ligne unique.

api_endpoint:
  type: string
  title: API Endpoint
  x-ui:
    widget: input

textarea

Zone de texte à plusieurs lignes pour un contenu plus long. Prend en charge une propriété facultative rows pour contrôler la hauteur.

message_body:
  type: string
  title: Message Body
  x-ui:
    widget: textarea
    rows: 4

checkbox

Case à cocher standard pour les valeurs booléennes.

send_notification:
  type: boolean
  title: Send Notification
  default: false
  x-ui:
    widget: checkbox

toggle

Basculez le commutateur pour les valeurs booléennes.

enable_logging:
  type: boolean
  title: Enable Logging
  default: true
  x-ui:
    widget: toggle

number

Champ d’entrée numérique. Utilisez minimum et maximum sur la propriété elle-même pour limiter la plage.

num_clusters:
  type: number
  title: Number of Clusters
  default: 3
  minimum: 1
  maximum: 100
  x-ui:
    widget: number

slider

Curseur visuel pour sélectionner des valeurs numériques dans une plage. Utilisez minimum et maximum sur la propriété pour définir la plage, puis stepx-ui pour contrôler l’incrément.

confidence_threshold:
  type: number
  title: Confidence Threshold
  default: 0.8
  minimum: 0
  maximum: 1
  x-ui:
    widget: slider
    step: 0.05

select

Liste déroulante sélection unique. Nécessite une optionsSource définition de l’emplacement d’où proviennent les valeurs de liste déroulante. Consultez les sources d’options.

aggregation_type:
  type: string
  title: Aggregation Type
  x-ui:
    widget: select
    optionsSource:
      type: static
      values: ['sum', 'avg', 'min', 'max', 'count']

multi-select

Liste déroulante à sélection multiple pour choisir plusieurs valeurs. Utilisez-la type: arrayitems: { type: string } sur la propriété. Nécessite un optionsSource. Consultez les sources d’options.

feature_columns:
  type: array
  title: Feature Columns
  items:
    type: string
  x-ui:
    widget: multi-select
    optionsSource:
      type: inputColumns
      port: input_data

expression

Sélecteur de colonne/expression qui permet aux utilisateurs de choisir une colonne à partir de données d’entrée ou d’écrire une expression SQL personnalisée. Définissez format: expression sur la propriété et spécifiez l’entrée port dans x-ui. Cela est utile :

• Lorsque l’utilisateur doit sélectionner une colonne dans les données d’entrée.
• Lorsque l’utilisateur peut vouloir écrire une expression SQL personnalisée.
• Pour les paramètres qui référencent des données dynamiques dans le pipeline.

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

Sources d’options

Pour select et multi-select les widgets, vous devez définir l’emplacement d’utilisation optionsSourcedes options de liste déroulante . Il existe deux sources : static (une liste fixe définie dans le YAML) et inputColumns (noms de colonnes à partir d’un port d’entrée).

Options statiques

Liste fixe de valeurs définies dans yaML.

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

Colonnes d’entrée

Remplit dynamiquement la liste déroulante avec des noms de colonnes à partir d’un port d’entrée.

optionsSource:
  type: inputColumns
  port: input_data

run_function

La propriété run_function vous permet d’incorporer du code Python directement dans la configuration YAML pour les opérateurs python-run-function. Cela élimine la nécessité d’inscrire une fonction de catalogue Unity distincte.

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

La run() fonction reçoit trois arguments :

• config: dictionnaire de valeurs de configuration définies par l’utilisateur dans l’interface utilisateur.
• inputs: dictionnaire mappant les noms de ports d’entrée aux DataFrames.
• spark: La SparkSession active.

La fonction doit retourner un nom de port de sortie de mappage de dictionnaire aux DataFrames. Les clés doivent correspondre exactement au name champ de chaque port de sortie défini dans ports.output. Par exemple, avec un port de sortie nommé out:

return {"out": result_df}

Avec plusieurs ports de sortie :

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

environment

La propriété environment spécifie l’environnement Python pour les opérateurs python-run-function. Utilisez-la pour épingler la version de l’environnement et déclarer des dépendances pip.

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

Exemples complets

UDF basée sur l’UC

Cet exemple définit un opérateur UDF basé sur le catalogue Unity qui calcule l’intérêt composé.

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

Python opérateur de fonction d’exécution

Cet exemple définit un python-run-function opérateur qui segmente les clients à l’aide du clustering K-Moyennes.

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'

Référence rapide

Propriétés racine requises

• schema : user-defined-operator-v0.1.0
• name: Nom d’affichage
• id: identificateur unique
• description: Que fait l’opérateur
• config: objet Schéma JSON
• type: uc-udf, ou uc-udtfpython-run-function
• version: chaîne de version définie par l’auteur

Propriétés racine facultatives

• ports: définitions de port d’entrée et de sortie
• run_function : code de Python inline (python-run-function uniquement)
• environment : environnement et dépendances Python (python-run-function uniquement)

Types de données de propriété config

string | boolean | number | integer | array | object

Widgets d’interface utilisateur

input | textarea | checkbox | toggle | number | slider | select | multi-select | expression

Sources d’options

static (valeurs fixes) | inputColumns (à partir du port d’entrée)

Mettre en forme les valeurs

expression | table_source | file_source | column_expressions | sort_expressions | aggregation_expressions | ai_function_expressions | is_preview | string[]