チュートリアル: K-means クラスタリング

このチュートリアルではscikit-learnを使用して K-means クラスタリングを実行するLakeflow Designer 用のPythonユーザー定義テーブル関数 (UDTF) 演算子を構築する方法を示します。UDTFは、データセット全体を処理する機械学習タスクに最適です。ユーザー定義演算子に関する背景情報については、「Lakeflow Designer のユーザー定義演算子」を参照してください。

概要

このチュートリアルでは、 Pythonを使用して UDTF ユーザー定義オペレーターを作成する手順を説明します。この演算子は、選択した列に対して K-Means クラスタリングを実行し、ユーザーが次の操作を実行できるようにします。

特徴として使用する列を選択します。
クラスターの数を指定します。
各行のクラスター割り当てを含むテーブルを取得します。

ステップ 1: UDTF ハンドラーパターンを理解する

UDTFは、3つの主要なメソッドを持つPythonクラスとして実装されています：

__init__() : 状態を初期化するために、行が処理される前に一度呼び出されます。
eval(row, ...) : データを累積するために、各入力行に対して呼び出されます。
terminate() : すべての行が処理された後に、結果を生成するために呼び出されます。

このパターンにより、UDTF は次のことが可能になります。

eval()通話中にすべてのデータポイントを収集する
K-Meansモデルをトレーニングする terminate()
行ごとにクラスター化された結果を生成します

__init__メソッドは、行が処理される前に1回実行されます。eval() 呼び出しの間で持続する状態を設定します。設定（ID 列、特徴量列、クラスター数）や、行データおよび特徴量のジューサーなどが含まれます。

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() 入力行ごとに1回呼び出されます。最初の行から構成をキャプチャし、各行とその数値フィーチャ値を累積します。

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)

注記

eval()のrow問題はPySpark Row オブジェクトです。アクセスしやすいように、 .asDict()を使用して辞書に変換します。

terminate() すべての行が蓄積された後にランします。収集した特徴量に対してK-Meansモデルをトレーニングし、入力行ごとに1つの出力行を生成して、各行のIDり当てられたクラスターをペアリングします。

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)

ステップ 2: オペレーターの YAML を作成する

YAML 構成は、 Lakeflow Designer でオペレーターがどのように表示されるかを定義します。この演算子の場合:

Number ( k ): 作成するクラスターの数
ウィジェットを選択 ( id_column ): 入力テーブルからの列がドロップダウンに入力されました
複数選択ウィジェット ( columns ): 複数のフィーチャ列の選択
optionsSource : 入力テーブルスキーマからドロップダウンを自動的に入力します
入力ポート : この演算子が表形式のデータを受け入れることを指定します

YAML
schema: user-defined-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

利用可能なすべてのプロパティ、データ型、ウィジェット、およびオプションの包括的なガイドについては、ユーザー定義演算子 YAML リファレンスを参照してください。

ステップ 3: Unity Catalog関数を作成する

YAML 構成と Python ハンドラークラスを 1 つのCREATE FUNCTIONステートメントに結合します。

main.example_output に関数を作成します。スキーマが存在しない場合は、最初にスキーマを作成します。

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: user-defined-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)
$$

ステップ 4: サンプルデータでテストする

テスト用のサンプル顧客データを作成します。

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

3つのクラスターを使用してK-Means UDTFをテストします。

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

この場合、クラスタリングの結果を元のデータと結合して、クラスタの割り当てを確認します。

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

ステップ 5: オペレーターを登録する

Lakeflow Designer でオペレーターを使用するには、それを.user_defined_operators.yamlファイルに追加して登録する必要があります。

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

注記

このファイルをユーザーフォルダ内に定義した場合、そのファイルはあなた以外には表示されません。詳細については、「オペレータを検出可能にする」を参照してください。

ステップ 6: 権限を設定する

この演算子を使用する必要があるユーザーにアクセス権を付与します。

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

Lakeflow Designer での演算子の使用

登録されると、オペレーターはLakeflow Designer に次のように表示されます：

データソースを接続するための入力ポート
行を一意に識別する列を選択するためのドロップダウン
クラスタリング機能として使用する列を選択するための複数選択
希望するクラスターの数を入力する数値

ユーザーは、コードを書かずに、顧客、製品、またはその他のデータを意味のあるグループに分割できます。

UDTF構築のヒント

__init__で状態を初期化する ：データを蓄積できるよう、空のリストや変数を設定します。
に蓄積eval ：まだ処理せず、データを収集するだけです。
terminateでの処理 : ここで本格的な作業が行われます。
yield「 」を使用した行の返却：結果を順次返します terminate
エッジケースの対応 ：クラスター数よりも行が少ない場合はどうなりますか？
型を明示的にする：UDTFの戻り値は入力型を参照できません。

概要​

ステップ 1: UDTF ハンドラー パターンを理解する​

ステップ 2: オペレーターの YAML を作成する​

ステップ 3: Unity Catalog関数を作成する​

ステップ 4: サンプルデータでテストする​

ステップ 5: オペレーターを登録する​

ステップ 6: 権限を設定する​

Lakeflow Designer での演算子の使用​

UDTF構築のヒント​

概要