Samouczek: klastrowanie metodą K-means

W tym samouczku utworzysz operator funkcji tabelarycznej zdefiniowanej przez użytkownika (UDTF) w języku Python dla narzędzia Lakeflow Designer, który wykonuje grupowanie metodą k-średnich za pomocą biblioteki scikit-learn. Funkcje UDTF dobrze nadają się do zadań uczenia maszynowego, które przetwarzają całe zbiory danych. Aby uzyskać informacje o niestandardowych operatorach, zobacz Niestandardowe operatory w Lakeflow Designer.

Overview

Ten samouczek przeprowadza Cię przez tworzenie niestandardowego operatora UDTF w Pythonie. Operator przeprowadza grupowanie metodą k-średnich w wybranych kolumnach, co pozwala użytkownikom na:

Wybierz kolumny, które mają być używane jako funkcje.
Określ liczbę klastrów.
Wróć do tabeli z przypisaniami klastra dla każdego wiersza.

Requirements

Dostęp do Lakeflow Designer w przestrzeni roboczej z Unity Catalog. Zobacz Wymagania wstępne.
USE CATALOG w katalogu main, oraz CREATE SCHEMA na main, lub USE SCHEMA oraz CREATE FUNCTION na istniejącym main.example_output schemacie. Aby użyć innego katalogu lub schematu, zastąp main.example_output w całym tekście. Zobacz odniesienie dotyczące uprawnień Unity Catalog.
Środowisko obliczeniowe mogące uruchamiać Unity Catalog Python UDTF. Zobacz Funkcje tabeli definiowane przez użytkownika (UDTF) w języku Python w katalogu Unity.

Krok 1. Omówienie wzorca obsługi UDTF

Funkcja UDTF jest implementowana jako klasa Python z trzema kluczowymi metodami:

__init__(): Wywołane raz, zanim jakiekolwiek wiersze zostaną przetworzone, aby zainicjować stan.
eval(row, ...): wywoływane dla każdego wiersza wejściowego w celu gromadzenia danych.
terminate(): Wywoływane po przetworzeniu wszystkich wierszy w celu uzyskania wyników.

Ten wzorzec umożliwia UDTF:

Zbieraj wszystkie dane podczas wywołań eval()
Trenuj model K-średnich w terminate()
Zwracaj wyniki klastrowane wiersz po wierszu

Metoda __init__ uruchamia się raz, zanim zostaną przetworzone jakiekolwiek wiersze. Inicjalizuje stan, który jest utrzymywany między wywołaniami eval(): konfigurację (kolumnę ID, kolumny cech, liczbę klastrów) oraz akumulatory danych wierszy i cech.

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() jest wywoływany raz dla każdego wiersza wejściowego. Pobiera konfigurację z pierwszego wiersza, a następnie gromadzi każdy wiersz oraz jego wartości cech numerycznych.

    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)

Note

Parametr row w pliku eval() jest obiektem wiersza PySpark. Użyj .asDict(), aby przekształcić je w słownik, co ułatwi dostęp.

terminate() jest wykonywane po zgromadzeniu wszystkich wierszy. Trenuje model K-Means na zebranych cechach i generuje jeden wiersz wyjściowy na każdy wiersz wejściowy, łącząc identyfikator każdego wiersza z przypisanym klastrem.

    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)

Krok 2. Tworzenie kodu YAML dla operatora

Konfiguracja YAML definiuje sposób wyświetlania operatora w projektancie lakeflow. Dla tego operatora:

Parametr liczbowy (k): liczba klastrów do utworzenia
Wybierz widżet (id_column): Lista rozwijana wypełniona kolumnami z tabeli wejściowej
Widżet wielokrotnego wyboru (columns): wybór wielu kolumn funkcji
optionsSource: Automatycznie wypełnia listy rozwijane ze schematu tabeli wejściowej
Port wejściowy: określa, że ten operator akceptuje dane tabelaryczne

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

Aby uzyskać wszystkie dostępne właściwości, typy danych, widżety i opcje, zobacz odniesienie do niestandardowego operatora YAML.

Krok 3. Utwórz funkcję w Unity Catalog

Połącz konfigurację YAML i klasę obsługi Python w jedną instrukcję CREATE FUNCTION.

Przykład tworzy funkcję w main.example_output. Najpierw stwórz schemat, jeśli nie istnieje:

CREATE SCHEMA IF NOT EXISTS main.example_output

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

Krok 4. Testowanie przy użyciu przykładowych danych

Tworzenie przykładowych danych klienta na potrzeby testowania:

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

Przetestuj K-Means UDTF na trzech klastrach:

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

W takim przypadku chcesz ponownie połączyć wyniki grupowania z oryginalnymi danymi, aby zobaczyć przypisania do klastrów:

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

Krok 5. Rejestrowanie operatora

Aby użyć operatora w programie Lakeflow Designer, należy go zarejestrować, dodając go do .custom_operators.yaml pliku:

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

Note

Jeśli zdefiniujesz ten plik w folderze użytkownika, zostanie on wyświetlony tylko dla Ciebie. Więcej informacji znajdziesz w artykule Zarejestruj i wdróż operator do Lakeflow Designer.

Krok 6. Konfigurowanie uprawnień

Udziel dostępu użytkownikom, którzy muszą używać tego operatora:

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

Krok 7: Użyj operatora w Lakeflow Designer

W przygotowaniu danych wizualnych otwórz menu Operatorów w lewym panelu i znajdź operatora klastrowania K-Means .
Przeciągnij operator na obszar roboczy i podłącz do niego wejście.
Kliknij dwukrotnie operatora, aby otworzyć panel konfiguracyjny, a następnie wypełnij jego pola.
Wybierz operatora, aby podglądać jego wyniki w panelu wyjściowym.

Operator posiada:

Port wejściowy do połączenia ze źródłem danych
Lista rozwijana umożliwiająca wybranie kolumny jednoznacznie identyfikującej wiersze
Pole wielokrotnego wyboru służące do wskazania kolumn używanych jako cechy grupowania
Liczba danych wejściowych dla żądanej liczby klastrów

Użytkownicy mogą podzielić klientów, produkty lub inne dane na znaczące grupy bez konieczności pisania kodu.

Wskazówki dotyczące tworzenia funkcji UDTF

Inicjowanie stanu w programie __init__: Konfigurowanie pustych list/zmiennych w celu gromadzenia danych
Gromadź w eval: Jeszcze nie przetwarzaj, tylko zbieraj dane
Proces w programie terminate: W tym miejscu odbywa się prawdziwa praca
yield
Obsługa przypadków brzegowych: Co zrobić, jeśli istnieje mniej wierszy niż klastry?
Jawnie określaj typy: typy zwracane przez UDTF nie mogą odwoływać się do typów wejściowych

Dodatkowe zasoby

Opinia

Czy ta strona była pomocna?

Ostatnia aktualizacja o 2026-10-26