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

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

概要

このチュートリアルでは、Python を使用して UDTF カスタム演算子を作成する手順を説明します。この演算子は、選択した列に対して K-Means クラスタリングを実行し、ユーザーが以下のことを行えるようにします。

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

要件

Unity Catalog が有効になっているワークスペース内の Lakeflow Designer へのアクセス。前提条件を参照してください。
USE CATALOG main カタログの CREATE SCHEMA と main、または既存の main.example_output スキーマの USE SCHEMA と CREATE FUNCTION。別のカタログまたはスキーマを使用するには、main.example_output全体を置き換えます。Unity Catalog の権限リファレンスを参照してください。
Unity Catalog Python UDTF を実行できるコンピュート。Unity Catalog の Python ユーザー定義テーブル関数 (UDTF)を参照してください。

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

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

__init__() ：状態を初期化するために行が処理される前に1回呼び出されます。
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: 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

すべての使用可能なプロパティ、データ型、ウィジェット、オプションについては、カスタムオペレーター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: 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)
$$

ステップ 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 でオペレーターを使用するには、それを.custom_operators.yamlファイルに追加して登録する必要があります。

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

注記

このファイルをユーザーフォルダに定義した場合、そのファイルは自分にのみ表示されます。情報については、 Lakeflow Designer へのオペレーターの登録とデプロイを参照してください。

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

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

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

ステップ 7: Lakeflow Designer で演算子を使用する

ビジュアルデータの準備で、左側のパネルの オペレーター メニューを開き、 K-Means Clustering オペレーターを見つけます。
演算子をキャンバスにドラッグし、入力を接続します。
オペレーターをダブルクリックして設定ウィンドウを開き、フィールドを設定します。
出力ペインで結果をプレビューするには、演算子を選択してください。

演算子には以下があります。

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

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

UDTF構築のヒント

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

概要​

要件​

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

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

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

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

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

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

ステップ 7: Lakeflow Designer で演算子を使用する​

UDTF構築のヒント​

その他のリソース​

概要

要件