Aller au contenu principal
Dernière mise à jour le

Écrire la logique pour les opérateurs définis par l’utilisateur 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 opérateur python-run-function, il s’agit d’une fonction Python run() en ligne. Pour un opérateur uc-udf ou uc-udtf, il s’agit d’une fonction Unity Catalog écrite en SQL ou en Python.

La rédaction de la logique est une étape de création d'un opérateur partagé défini par l'utilisateur. Après avoir écrit et testé la logique, vous décrivez l'opérateur au format YAML pour que Lakeflow Designer sache comment l'afficher dans l'interface utilisateur, puis vous l'enregistrez pour qu'il apparaisse dans le volet Opérateurs . Pour voir toutes les étapes d'un opérateur, suivez un tutoriel comme le Tutoriel : Opérateur de calcul d'intérêts composés.

Écrire un opérateur python-run-function​

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

Python
def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config : Valeurs configurées par l'utilisateur à partir de l'interface utilisateur, classées par nom de propriété.
  • inputs : DataFrames d’entrée, indexés par le port d’entrée name.
  • spark : La SparkSession active.
  • Retourne : dictionnaire mappant les valeurs du port de sortie name vers des DataFrames.

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

Python
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 notebook avec des données d'exemple :

Python
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 nécessite des packages pip externes, ajoutez le champ environment au YAML :

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 Unity Catalog en SQL ou en Python. Les fonctions Python sont encapsulées dans une instruction SQL CREATE FUNCTION :

Fonction SQL :

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

Fonction Python (encapsulée en SQL) :

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 maintenir leur état sur toutes les lignes. Utilisez uc-udf pour les transformations au niveau de la colonne et uc-udtf pour les opérations telles que le clustering ML ou l’agrégation.

Une classe de gestionnaire UDTF définit eval(), qui s'exécute pour chaque ligne en entrée. Facultativement, elle définit également __init__() pour configurer l'état et terminate() pour produire les résultats finaux après toutes les lignes. Voir Fonctions de table définies par l'utilisateur (UDTF) Python dans Unity Catalog.

Python
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 renvoi UDTF doivent avoir des types fixes et explicites. Vous ne pouvez pas référencer des types de colonnes d'entrée dans la configuration de retour.

Sauter les effets secondaires pendant les prévisualisations​

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

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

Lakeflow Designer définit automatiquement cette valeur sur true pendant l'aperçu. Vérifiez-la dans votre logique pour ignorer les effets secondaires :

Python
# In a python-run-function
if config.get("is_preview"):
    return {"out": inputs["in"]}
SQL
-- In a UC function (SQL)
CASE WHEN is_preview THEN 'preview' ELSE /* actual work */ END

Appelez des APIs externes avec des connexions HTTP Unity Catalog​

Pour les opérateurs définis par l’utilisateur basés sur les fonctions SQL Unity Catalog qui appellent des APIs externes, utilisez les connexions HTTP Unity Catalog pour stocker les informations d’identification de manière sécurisée :

SQL
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 de détails, consultez Connecter à des services HTTP externes.

Accéder aux ressources du workspace avec WorkspaceClient​

Pour les opérateurs python-run-function, vous pouvez utiliser Databricks WorkspaceClient pour accéder aux ressources du workspace et aux APIs externes :

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

Ressources supplémentaires​

Sur cette page