Didacticiel : clustering K-means

Dans ce tutoriel, vous allez créer un opérateur de fonction de table définie par l'utilisateur (UDTF) Python pour Lakeflow Designer qui exécute le clustering K-means avec scikit-learn. Les UDTF sont bien adaptés aux tâches de machine learning qui traitent des datasets entiers. Pour plus d'informations sur les opérateurs personnalisés, voir Opérateurs personnalisés dans Lakeflow Designer.

Présentation

Ce tutoriel vous guide dans la création d'un opérateur personnalisé UDTF à l'aide de Python. L'opérateur effectue un clustering K-Means sur les colonnes sélectionnées, permettant aux utilisateurs de :

Choisissez les colonnes à utiliser comme caractéristiques.
Spécifiez le nombre de clusters.
Obtenez un tableau avec les affectations de clusters pour chaque ligne.

Conditions requises

Accès à Lakeflow Designer dans un workspace activé pour Unity Catalog. Voir Conditions préalables.
USE CATALOG sur le catalogue main, et CREATE SCHEMA sur main, ou USE SCHEMA et CREATE FUNCTION sur un schéma main.example_output existant. Pour utiliser un autre catalogue ou schéma, remplacez main.example_output partout. Voir Référence des privilèges Unity Catalog.
Compute capable d'exécuter des UDTF Python Unity Catalog. Consultez Fonctions de table définies par l'utilisateur (UDTF) Python dans Unity Catalog.

Étape 1 : Comprendre le modèle de gestionnaire UDTF

Une UDTF est implémentée comme une classe Python avec trois méthodes clés :

__init__() : Appelé une fois, avant que les lignes ne soient traitées, pour initialiser l’état.
eval(row, ...) ** ** : Appelée pour chaque ligne d'entrée afin d'accumuler les données.
terminate() : Appélé après que toutes les lignes ont été traitées pour produire des résultats.

Ce modèle permet à l'UDTF de :

Collectez tous les points de données pendant eval() appels
Entraîner le modèle K-Means dans terminate()
Produire des résultats en cluster ligne par ligne

La méthode __init__ s’exécute une seule fois, avant le traitement de toute ligne. Elle configure l’état qui persiste entre les appels eval() : la configuration (colonne d’ID, colonnes de caractéristiques, nombre de clusters) et les accumulateurs pour les données de ligne et les caractéristiques.

Python
class SklearnKMeans:
    def __init__(self):
        # State set here persists across eval() calls for use in terminate()
        self.id_col = None
        self.feature_cols = None
        self.k = None
        self.rows = []
        self.features = []

eval() est appelé une fois pour chaque ligne d’entrée. Il capture la configuration de la première ligne, puis accumule chaque ligne et ses valeurs de fonctionnalité numériques.

Python
    def eval(self, row, id_column, columns, k):
        if self.id_col is None:
            self.id_col = id_column
        if self.feature_cols is None:
            self.feature_cols = columns
        if self.k is None:
            self.k = max(1, int(k))

        row_dict = row.asDict(recursive=False)
        self.rows.append(row_dict)

        feats = []
        for c in self.feature_cols:
            v = row_dict.get(c)
            if v is None:
                v = 0.0
            feats.append(float(v))
        self.features.append(feats)

remarque

Le row paramètre dans eval() est un objet PySpark Row. Utilisez .asDict() pour le convertir en dictionnaire pour un accès plus facile.

terminate() s’exécute après l’accumulation de toutes les lignes. Il entraîne le modèle K-Means sur les caractéristiques collectées et produit une ligne de sortie par ligne d’entrée, en associant l’ID de chaque ligne à son cluster assigné.

Python
    def terminate(self):
        import numpy as np
        from sklearn.cluster import KMeans

        if not self.rows:
            return

        X = np.asarray(self.features, dtype=float)
        n_samples = X.shape[0]
        n_clusters = min(self.k, n_samples)

        model = KMeans(
            n_clusters=n_clusters,
            n_init=10,
            random_state=42
        )
        labels = model.fit_predict(X)

        for row_dict, label in zip(self.rows, labels):
            yield str(row_dict[self.id_col]), int(label)

Étape 2 : Créez le YAML pour l'opérateur

La configuration YAML définit la manière dont l'opérateur apparaît dans Lakeflow Designer. Pour cet opérateur :

Paramètre numérique (k) : nombre de clusters à créer
Sélectionner le widget (id_column) : Liste déroulante renseignée avec les colonnes de la table d'entrée
**Widget columnsà sélection multiple** () : sélection de plusieurs colonnes de fonctionnalités.
optionsSource : Remplit automatiquement les listes déroulantes à partir du schéma de la table d'entrée
Port d'entrée : Spécifie que cet opérateur accepte les données tabulaires

YAML
schema: custom-operator-v0.1.0
type: uc-udtf
name: K-Means Clustering
id: kmeans
version: '1.0.0'
description: Perform K-Means clustering on selected columns
config:
  type: object
  properties:
    k:
      type: number
      title: Number of Clusters
      default: 3
      minimum: 1
      maximum: 100
      x-ui:
        widget: number
    id_column:
      type: string
      title: ID Column
      x-ui:
        widget: select
        optionsSource:
          type: inputColumns
          port: input_data
    columns:
      type: array
      items:
        type: string
      title: Feature Columns
      x-ui:
        widget: multi-select
        optionsSource:
          type: inputColumns
          port: input_data
  required:
    - k
    - id_column
    - columns
  additionalProperties: false
ports:
  input:
    - name: input_data
      title: Input Data
  output:
    - name: output
      title: Clustered Data

Pour obtenir la liste complète des propriétés, types de données, widgets et options disponibles, consultez la référence YAML de l'opérateur personnalisé.

Étape 3 : créez la fonction Unity Catalog

Combinez la configuration YAML et la classe du gestionnaire Python en une seule instruction CREATE FUNCTION.

L’exemple crée la fonction dans main.example_output. Créez d’abord le schéma s’il n’existe pas :

SQL
CREATE SCHEMA IF NOT EXISTS main.example_output

SQL
CREATE OR REPLACE FUNCTION main.example_output.k_means(
    input_data TABLE,
    id_column STRING,
    columns ARRAY<STRING>,
    k INT
)
RETURNS TABLE (
    id STRING,
    cluster_id INT
)
LANGUAGE PYTHON
HANDLER 'SklearnKMeans'
AS $$
"""
schema: custom-operator-v0.1.0
type: uc-udtf
name: K-Means Clustering
id: kmeans
version: "1.0.0"
description: Perform K-Means clustering on selected columns
config:
  type: object
  properties:
    k:
      type: number
      title: Number of Clusters
      default: 3
      minimum: 1
      maximum: 100
      x-ui:
        widget: number
    id_column:
      type: string
      title: ID Column
      x-ui:
        widget: select
        optionsSource:
          type: inputColumns
          port: input_data
    columns:
      type: array
      items:
        type: string
      title: Feature Columns
      x-ui:
        widget: multi-select
        optionsSource:
          type: inputColumns
          port: input_data
  required:
    - k
    - id_column
    - columns
  additionalProperties: false
ports:
  input:
    - name: input_data
      title: Input Data
  output:
    - name: output
      title: Clustered Data
"""

class SklearnKMeans:
    def __init__(self):
        self.id_col = None
        self.feature_cols = None
        self.k = None
        self.rows = []
        self.features = []

    def eval(self, row, id_column, columns, k):
        if self.id_col is None:
            self.id_col = id_column
        if self.feature_cols is None:
            self.feature_cols = columns
        if self.k is None:
            self.k = max(1, int(k))

        row_dict = row.asDict(recursive=False)
        self.rows.append(row_dict)

        feats = []
        for c in self.feature_cols:
            v = row_dict.get(c)
            if v is None:
                v = 0.0
            feats.append(float(v))
        self.features.append(feats)

    def terminate(self):
        import numpy as np
        from sklearn.cluster import KMeans

        if not self.rows:
            return

        X = np.asarray(self.features, dtype=float)
        n_samples = X.shape[0]
        n_clusters = min(self.k, n_samples)

        model = KMeans(
            n_clusters=n_clusters,
            n_init=10,
            random_state=42
        )
        labels = model.fit_predict(X)

        for row_dict, label in zip(self.rows, labels):
            yield str(row_dict[self.id_col]), int(label)
$$

Étape 4 : Tester avec des exemples de données

Créer un échantillon de données client pour les tests :

SQL
CREATE OR REPLACE TEMP VIEW customers AS
SELECT * FROM VALUES
    ('C001', 25, 35000, 20),
    ('C002', 45, 85000, 80),
    ('C003', 35, 55000, 50),
    ('C004', 50, 95000, 90),
    ('C005', 23, 30000, 15),
    ('C006', 40, 75000, 70),
    ('C007', 60, 100000, 95),
    ('C008', 30, 45000, 40)
AS t(customer_id, age, annual_income, spending_score);

Testez l'UDTF K-Means avec trois clusters :

SQL
SELECT * FROM main.example_output.k_means(
    input_data => TABLE(SELECT * FROM customers) WITH SINGLE PARTITION,
    k => 3,
    id_column => 'customer_id',
    columns => array('age', 'annual_income', 'spending_score')
)

Dans ce cas, vous souhaitez joindre les résultats du clustering avec les données d'origine pour voir les attributions de clusters :

SQL
SELECT
  c.*,
  k.cluster_id
FROM customers c
INNER JOIN main.example_output.k_means(
    input_data => TABLE(SELECT * FROM customers) WITH SINGLE PARTITION,
    k => 3,
    id_column => 'customer_id',
    columns => array('age', 'annual_income', 'spending_score')
) k
ON c.customer_id = k.id
ORDER BY k.cluster_id, c.customer_id

Étape 5 : Enregistrez l'opérateur

Pour utiliser l'opérateur dans Lakeflow Designer, vous devez l'enregistrer en l'ajoutant à votre fichier .custom_operators.yaml :

YAML
operators:
  - catalog: main
    schema: example_output
    functionName: k_means

remarque

Si vous définissez ce fichier dans votre dossier utilisateur, il n'apparaît que pour vous. Pour plus d'informations, consultez Enregistrer et déployer votre opérateur sur Lakeflow Designer.

Étape 6 : Configurez les autorisations

Accordez l’accès aux utilisateurs qui doivent utiliser cet opérateur :

SQL
GRANT USE SCHEMA ON SCHEMA main.example_output TO `<user>`;
GRANT EXECUTE ON FUNCTION main.example_output.k_means TO `<user>`;

Étape 7 : Utiliser l'opérateur dans Lakeflow Designer

Dans une préparation de données visuelles, ouvrez le menu Opérateurs dans le panneau gauche et recherchez l’opérateur Clustering k-moyennes .
Faites glisser l'opérateur sur la toile et connectez une entrée à celui-ci.
Double-cliquez sur l’opérateur pour ouvrir son panneau de configuration, puis définissez ses champs.
Sélectionnez l’opérateur pour prévisualiser ses résultats dans le volet de sortie.

L'opérateur dispose de :

Un port d'entrée pour connecter votre source de données
Une liste déroulante pour sélectionner la colonne qui identifie de manière unique les lignes.
Une sélection multiple pour choisir les colonnes à utiliser comme fonctionnalités de clustering
Une entrée numérique pour le nombre de clusters souhaité

Les utilisateurs peuvent segmenter les clients, les produits ou toute autre donnée en groupes significatifs sans écrire de code.

Conseils pour la création d’UDTF

Initialiser l'état dans __init__ : Configurez des listes/variables vides pour accumuler des données.
Accumuler dans eval : Ne traitez pas encore, collectez simplement les données
Traiter en terminate : c’est là que le vrai travail a lieu
Utilisez yield pour retourner les lignes : Retournez les résultats un par un de terminate
Gérer les cas limites : Que faire s'il y a moins de lignes que de clusters ?
Gardez les types explicites : les retours UDTF ne peuvent pas référencer les types d’entrée

Présentation​

Conditions requises​

Étape 1 : Comprendre le modèle de gestionnaire UDTF​

Étape 2 : Créez le YAML pour l'opérateur​

Étape 3 : créez la fonction Unity Catalog​

Étape 4 : Tester avec des exemples de données​

Étape 5 : Enregistrez l'opérateur​

Étape 6 : Configurez les autorisations​

Étape 7 : Utiliser l'opérateur dans Lakeflow Designer​

Conseils pour la création d’UDTF​

Ressources supplémentaires​