Tutorial: clusteringK-means

Neste tutorial, você criará um operador de função de tabela definida pelo usuário (UDTF) em Python para o Lakeflow Designer que faz a execução do clustering K-means com scikit-learn. As UDTFs são bem adequadas para tarefas de machine learning que processam conjuntos de dados inteiros. Para obter mais informações sobre operadores personalizados, consulte Operadores personalizados no Lakeflow Designer.

Visão geral

Este tutorial orienta-o na criação de um operador personalizado UDTF utilizando Python. O operador realiza o agrupamento K-Means nas colunas selecionadas, permitindo aos utilizadores:

Escolha quais colunas usar como recurso.
Especifique o número de clusters.
Obtenha uma tabela com as atribuições de cluster para cada linha.

Requisitos

Acesso ao Lakeflow Designer em um workspace habilitado para Unity Catalog. Consulte Pré-requisitos.
USE CATALOG no catálogo main, e CREATE SCHEMA no main, ou USE SCHEMA e CREATE FUNCTION em um esquema existente main.example_output. Para usar um catálogo ou esquema diferente, substitua main.example_output por toda parte. Consulte Referência de privilégios do Unity Catalog.
Compute que pode executar UDTFs Python do Unity Catalog. Consulte Funções de tabela definidas pelo usuário Python (UDTFs) no Unity Catalog.

Passo 1: Compreender o padrão de manipulador UDTF

Uma UDTF é implementada como uma classe Python com três métodos chave:

__init__() : Chamado uma vez, antes de qualquer linha ser processada para inicializar o estado.
eval(row, ...) ** **: Chamado para cada linha de entrada para acumular dados.
terminate() : Chamado após todas as linhas serem processadas para produzir resultados.

Esse padrão permite que a UDTF:

Coletar todos os pontos de dados durante as chamadas eval()
ensinar o modelo K-Means em terminate()
Obtenha resultados agrupados linha por linha.

O método __init__ é executado uma vez, antes que qualquer linha seja processada. Ele define o estado que persiste através das chamadas de eval(): a configuração (coluna ID, colunas de recurso, contagem de clusters) e os acumuladores para dados de linha e recursos.

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() é chamado uma vez para cada linha de entrada. Ela captura a configuração da primeira linha, depois acumula cada linha e seus valores de recurso numéricos.

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)

nota

O parâmetro row em eval() é um objeto Row do PySpark. Use .asDict() para convertê-lo em um dicionário para facilitar o acesso.

terminate() executa após todas as linhas serem acumuladas. Ele treina o modelo K-Means nos recursos coletados e gera uma linha de saída por linha de entrada, pareando o ID de cada linha com seu cluster atribuído.

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)

o passo 2: Crie o YAML para o operador

A configuração YAML define como o operador aparece no LakeFlow Designer. Para esta operadora:

Parâmetro numérico (k): Número de clusters a serem criados
Widget de seleção (id_column): menu suspenso preenchido com colunas da tabela de entrada
Widget de seleção múltipla (columns): Seleção de várias colunas de recursos
optionsSource : Preenche automaticamente a lista suspensa a partir do esquema da tabela de entrada
Porta de entrada : Especifica que este operador aceita dados tabulares.

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

Para todas as propriedades, tipos de dados, widgets e opções disponíveis, consulte Referência de YAML de operador customizado.

o passo 3: Criar a função Unity Catalog

Combine a configuração YAML e a classe manipuladora Python em uma única declaração CREATE FUNCTION .

O exemplo cria a função em main.example_output. Crie o esquema primeiro se ele não existir:

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

a etapa 4: Teste com dados de amostra

Crie dados de amostra do cliente para teste:

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

Teste o UDTF K-Means com três 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')
)

Nesse caso, você deseja join os resultados clustering com os dados originais para visualizar as atribuições cluster :

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

o passo 5: registrar a operadora

Para usar o operador no LakeFlow Designer, você deve registrá-lo, adicionando-o ao seu arquivo .custom_operators.yaml :

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

nota

Se você definir esse arquivo em sua pasta de usuário, ele aparecerá apenas para você. Para obter mais informações, consulte Registrar e implantar seu operador no Lakeflow Designer.

o passo 6: Configurar permissões

Conceda acesso aos usuários que precisam usar esta operadora:

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

O passo 7: Usar o operador no Lakeflow Designer

Em uma preparação de dados visuais, abra o menu Operadores no painel esquerdo e localize o operador Agrupamento K-Means .
Arraste o operador para a tela e conecte uma entrada a ele.
Clique duas vezes no operador para abrir seu painel de configuração e, em seguida, defina seus campos.
Selecione o operador para visualizar os resultados no painel de saída.

O operador tem:

Uma porta de entrada para conectar sua fonte de dados.
Um dropdown para selecionar qual coluna identifica exclusivamente as linhas.
Uma seleção múltipla para escolher quais colunas usar como recurso clustering .
Um número de entrada para a quantidade desejada de clusters.

Os usuários podem segmentar clientes, produtos ou quaisquer outros dados em grupos significativos sem precisar escrever código.

Dicas para criar UDTFs

Inicializar estado em __init__ : Configurar listas/variáveis vazias para acumular dados
Acumular em eval : Ainda não processar, apenas coletar dados
Processo em terminate : É aqui que o verdadeiro trabalho acontece
Use yield para retornar linhas : Retornar resultados um por vez de terminate
Tratar casos extremos : e se houver menos linhas do que clusters?
Mantenha os tipos explícitos: retornos de UDTF não podem referenciar tipos de entrada

Visão geral​

Requisitos​

Passo 1: Compreender o padrão de manipulador UDTF​

o passo 2: Crie o YAML para o operador​

o passo 3: Criar a função Unity Catalog​

a etapa 4: Teste com dados de amostra​

o passo 5: registrar a operadora​

o passo 6: Configurar permissões​

O passo 7: Usar o operador no Lakeflow Designer​

Dicas para criar UDTFs​

Outros recursos​