Écrire la logique pour les opérateurs définis par l’utilisateur de Lakeflow Designer

La logique de l’opérateur est le code qui s’exécute lorsqu’un opérateur défini par l’utilisateur s’exécute. Pour un python-run-function opérateur, c'est une fonction Python run() en ligne. Pour un opérateur uc-udf ou uc-udtf, c'est une fonction du catalogue Unity écrite en SQL ou Python.

Écrire la logique est une étape pour construire un opérateur partagé défini par l’utilisateur. Après avoir écrit et testé la logique, vous décrivez l’opérateur en YAML pour que Lakeflow Designer sache comment l’afficher dans l’interface, puis vous l’enregistrez pour qu’il apparaisse dans le panneau Opérateurs . Pour voir toutes les étapes réunies pour un opérateur, suivez un tutoriel tel que Tutoriel : opérateur de calcul des intérêts composés.

Écrire un opérateur python-run-function

Chaque python-run-function opérateur doit définir une run() fonction :

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: Valeurs configurées par l’utilisateur dans l’interface utilisateur, indexées par nom de propriété.
  • inputs: DataFrames d’entrée, indexées par port d'entrée name.
  • spark : Ssession Spark active.
  • Renvoie : dictionnaire qui associe les valeurs du port de sortie name à des DataFrames.

L’exemple suivant filtre les lignes d’un DataFrame d’entrée :

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

Pour tester la fonction avant de l’enregistrer, appelez-la dans un carnet avec des données d’exemple :

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

Si votre opérateur requiert des packages pip externes, ajoutez le environment champ au YAML :

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

Écrire un opérateur UDF ou UDTF

Vous pouvez écrire des fonctions du catalogue Unity en SQL ou en Python. Les fonctions Python sont intégrées dans une instruction SQL CREATE FUNCTION :

Fonction 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 fonction (encapsulée dans 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)
$$;

Les UDF traitent une seule valeur à la fois et renvoient une valeur calculée. Les UDTF traitent les tables ligne par ligne et peuvent conserver l’état d’une ligne à l’autre. Utiliser uc-udf pour les transformations au niveau des colonnes et uc-udtf pour les opérations telles que le clustering ml ou l’agrégation.

Une classe gestionnaire UDTF définit eval(), qui s’exécute pour chaque ligne d’entrée. En option, il définit également __init__() pour configurer l’état et terminate() pour produire les résultats finaux après le traitement de toutes les lignes. Consultez Fonctions de table Python définies par l’utilisateur dans Unity Catalog.

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

Remarque

Les tables de retour UDTF doivent avoir des types fixes et explicites. Vous ne pouvez pas référencer les types de colonnes d’entrée dans la configuration de retour.

Sauter les effets secondaires pendant les aperçus

Lakeflow Designer prend en charge les aperçus en mode de conception. Pour les opérateurs qui appellent des API externes ou écrivent dans des systèmes externes, ajoutez une is_preview propriété de configuration pour que vous puissiez ignorer les effets secondaires pendant la préversion. Les utilisateurs cliquent sur Exécuter pour exécuter l’opérateur avec ses effets de bord. L'opérateur Python intégré n'a pas besoin de cette propriété, car il lit config["is_preview"] automatiquement. Voir le mode de prévisualisation Python.

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

Lakeflow Designer définit automatiquement cette valeur true pendant la préversion. Vérifiez-le dans votre logique pour ignorer les effets secondaires :

# 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

Appelez des API externes avec des connexions HTTP du catalogue Unity

Pour les opérateurs définis par l’utilisateur basés sur des fonctions SQL Unity Catalog qui appellent des API externes, utilisez les connexions HTTP Unity Catalog pour stocker en toute sécurité les identifiants :

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

Utilisez ensuite la connexion dans votre UDF SQL avec la fonction http_request(). Pour plus d’informations, consultez Se connecter aux services HTTP externes.

Accédez aux ressources de l’espace de travail avec WorkspaceClient

Pour les opérateurs python-run-function, vous pouvez utiliser le Azure Databricks WorkspaceClient pour accéder aux ressources de l’espace de travail et aux API externes :

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

Ressources supplémentaires

  • Dernière mise à jour :