Fonctions de table définies par l'utilisateur Python (UDTFs) dans Unity Catalog

info

Aperçu

L'enregistrement des UDTF Python dans Unity Catalog est en aperçu public.

Une fonction de table définie par l'utilisateur (UDTF) Unity Catalog enregistre des fonctions qui renvoient des tables complètes au lieu de valeurs scalaires. Contrairement aux fonctions scalaires qui renvoient une seule valeur de résultat à chaque appel, les UDTF sont appelées dans la clause FROM d'une instruction SQL et peuvent renvoyer plusieurs lignes et colonnes.

Les UDTF sont particulièrement utiles pour :

• Transformation des tableaux ou des structures de données complexes en plusieurs lignes
• Intégration d'APIs ou de services externes dans les workflows SQL
• Mise en œuvre de logique de génération ou d'enrichissement de données personnalisée
• Traitement des données qui nécessitent des opérations avec état sur les lignes

Chaque appel UDTF accepte zéro ou plusieurs arguments. Ces arguments peuvent être des expressions scalaires ou des arguments de table représentant des tables d'entrée entières.

Les UDTF peuvent être enregistrées de deux manières :

• Unity Catalog : Enregistrez l'UDTF en tant qu'objet gouverné dans Unity Catalog.
• Limité à la session : Enregistrez-vous sur le SparkSession local, isolé du Notebook ou du Job actuel. Consultez les fonctions de table définies par l'utilisateur Python (UDTFs).

Exigences

Les UDTF Python Unity Catalog sont pris en charge sur les types de compute suivants :

• Notebooks et Jobs Serverless
• Compute classique avec mode d'accès standard (Databricks Runtime 17.1 et versions supérieures)
• SQL Warehouse (Serverless ou Pro)

Créer une UDTF dans Unity Catalog

Utilisez le DDL SQL pour créer une UDTF régie dans Unity Catalog. Les UDTF sont invoquées à l'aide de la clause FROM d'une instruction SQL.

SQL

CREATE OR REPLACE FUNCTION square_numbers(start INT, end INT)
RETURNS TABLE (num INT, squared INT)
LANGUAGE PYTHON
HANDLER 'SquareNumbers'
DETERMINISTIC
AS $$
class SquareNumbers:
    """
    Basic UDTF that computes a sequence of integers
    and includes the square of each number in the range.
    """
    def eval(self, start: int, end: int):
        for num in range(start, end + 1):
            yield (num, num * num)
$$;

SELECT * FROM square_numbers(1, 5);

Output

+-----+---------+
| num | squared |
+-----+---------+
| 1   | 1       |
| 2   | 4       |
| 3   | 9       |
| 4   | 16      |
| 5   | 25      |
+-----+---------+

Databricks implémente les UDTF Python en tant que classes Python avec une méthode eval obligatoire qui produit des lignes de sortie.

Arguments de la table

remarque

Les arguments TABLE sont pris en charge dans Databricks Runtime 17.2 et versions ultérieures.

Les UDTF peuvent accepter des tables entières comme arguments d'entrée, ce qui permet des Transformations et des agrégations complexes et avec état.

Méthodes de cycle de vieeval() et terminate()

Les arguments de table dans les fonctions UDTF utilisent les fonctions suivantes pour traiter chaque ligne :

• eval(): Appelé une fois pour chaque ligne dans la table d'entrée. C'est la méthode de traitement principale et elle est requise.
• terminate(): Appelé une fois à la fin de chaque partition, après que toutes les lignes ont été traitées par eval(). Utilisez cette méthode pour obtenir des résultats agrégés finaux ou effectuer des opérations de nettoyage. Cette méthode est facultative mais essentielle pour les opérations avec état, telles que les agrégations, les comptages ou le traitement par batch.

Pour plus d'informations sur les méthodes eval() et terminate(), consultez la documentation Apache Spark : Python UDTF.

Modèles d'accès aux lignes

eval() reçoit des lignes des arguments TABLE en tant qu'objets pyspark.sql.Row. Vous pouvez accéder aux valeurs par nom de colonne (row['id'], row['name']) ou par index (row[0], row[1]).

• Flexibilité de schéma : Déclarez les arguments TABLE sans définitions de schéma (par exemple, data TABLE, t TABLE). La fonction accepte toute structure de table, votre code doit donc valider que les colonnes requises existent.

Voir Exemple : Faire correspondre des adresses IP à des blocs réseau CIDR et Exemple : Légendage d'images par batch à l'aide des Endpoint de vision Databricks.

Calculer un schéma de sortie dynamique (UDTF polymorphes)

remarque

Les UC UDTFs polymorphes nécessitent Databricks Runtime 18.1 et versions ultérieures.

Une UDTF polymorphe détermine son schéma de sortie dynamiquement au moment de la query à l'aide d'une méthode statique analyze(), plutôt que de déclarer les colonnes à l'avance. Pour en créer une, utilisez RETURNS TABLE sans définitions de colonne et définissez une méthode analyze() sur la classe de gestionnaire.

L'exemple suivant extrait les champs spécifiés par l'appelant d'une chaîne JSON, retournant différentes colonnes selon l'argument fields :

SQL

CREATE OR REPLACE FUNCTION extract_fields(json_str STRING, fields STRING)
RETURNS TABLE
LANGUAGE PYTHON
HANDLER 'ExtractFields'
AS $$
class ExtractFields:
    @staticmethod
    def analyze(json_str, fields):

        # Build the output schema from the requested field names
        from pyspark.sql.types import StructType, StructField, StringType
        from pyspark.sql.udtf import AnalyzeResult
        col_names = [f.strip() for f in fields.value.split(",")]
        return AnalyzeResult(
            StructType([StructField(name, StringType()) for name in col_names])
        )

    def eval(self, json_str: str, fields: str):
        # Parse the JSON and yield only the requested fields
        import json
        data = json.loads(json_str)
        col_names = [f.strip() for f in fields.split(",")]
        yield tuple(data.get(name) for name in col_names)
$$;

-- Extract the name and city
SELECT * FROM extract_fields(
  '{"name": "Alice", "age": 30, "city": "Seattle"}',
  'name, city'
);

Output

+-------+---------+
| name  | city    |
+-------+---------+
| Alice | Seattle |
+-------+---------+

Définir la méthode analyze

La classe de gestionnaire doit inclure une méthode @staticmethod nommée analyze qui accepte les mêmes arguments que le UDTF et renvoie un AnalyzeResult décrivant le schéma de sortie. Databricks appelle analyze() au moment de la planification de la query pour résoudre le schéma avant d'exécuter la fonction.

Chaque paramètre de analyze est une instance de la classe AnalyzeArgument :

Champ	Description
dataType	Le type de l’argument d’entrée en tant que DataType. Pour les arguments de table d'entrée, il s'agit d'un StructType représentant les colonnes de la table.
value	La valeur de l'argument d'entrée en tant que Optional[Any]. Ceci est None pour les arguments de table ou les expressions non constantes.
isTable	Si l'argument d'entrée est un argument de table en tant que BooleanType.
isConstantExpression	Si l'argument d'entrée est une expression à réduction constante en tant que BooleanType.

Champ	Description
dataType	Le type de l’argument d’entrée en tant que DataType. Pour les arguments de table d'entrée, il s'agit d'un StructType représentant les colonnes de la table.
value	La valeur de l'argument d'entrée en tant que Optional[Any]. Ceci est None pour les arguments de table ou les expressions non constantes.
isTable	Si l'argument d'entrée est un argument de table en tant que BooleanType.
isConstantExpression	Si l'argument d'entrée est une expression à réduction constante en tant que BooleanType.

La méthode analyze renvoie une instance de la classe AnalyzeResult :

Champ	Description
schema	Le schéma de la table de résultats sous la forme d'un StructType.
withSinglePartition	Si True, envoie toutes les lignes d'entrée à la même instance de classe UDTF.
partitionBy	Si non vide, partitionne les lignes d'entrée par les expressions spécifiées de sorte que chaque combinaison unique soit traitée par une instance UDTF distincte.
orderBy	Si non vide, spécifie un ordre des lignes au sein de chaque partition.
select	S'il n'est pas vide, spécifie quelles colonnes de l'argument TABLE d'entrée la UDTF reçoit.

Champ	Description
schema	Le schéma de la table de résultats sous la forme d'un StructType.
withSinglePartition	Si True, envoie toutes les lignes d'entrée à la même instance de classe UDTF.
partitionBy	Si non vide, partitionne les lignes d'entrée par les expressions spécifiées de sorte que chaque combinaison unique soit traitée par une instance UDTF distincte.
orderBy	Si non vide, spécifie un ordre des lignes au sein de chaque partition.
select	S'il n'est pas vide, spécifie quelles colonnes de l'argument TABLE d'entrée la UDTF reçoit.

attention

Pour les UDTF polymorphes de Unity Catalog, vous devez placer toutes les importations à l'intérieur du corps de la méthode analyze(). Les importations de haut niveau ne sont pas disponibles dans l'environnement sandbox de Unity Catalog.

Transférer l’état de analyze vers eval

La méthode analyze s'exécute une fois au moment de la planification de la query, de sorte que vous pouvez l'utiliser pour prétraiter les arguments constants, analyser les configurations ou construire des recherches. Pour transmettre ces résultats à eval, créez une sous-classe @dataclass de AnalyzeResult avec des champs personnalisés, renvoyez-la depuis analyze et acceptez-la dans la méthode __init__. Cela évite de répéter un travail coûteux pour chaque ligne.

L'exemple suivant résout un code de langue en nom de langue complet une fois dans analyze et le transmet, afin que eval puisse taguer chaque ligne sans répéter la recherche :

SQL

CREATE OR REPLACE FUNCTION tag_language(t TABLE, lang_code STRING)
RETURNS TABLE
LANGUAGE PYTHON
HANDLER 'TagLanguage'
AS $$
class TagLanguage:
    @staticmethod
    def analyze(t, lang_code):
        from dataclasses import dataclass
        from pyspark.sql.types import StructType, StructField, StringType
        from pyspark.sql.udtf import AnalyzeResult

        @dataclass
        class LangResult(AnalyzeResult):
            language: str = ""

        # Resolve the language code to a full name once during planning
        languages = {"en": "English", "es": "Spanish", "fr": "French", "de": "German"}
        return LangResult(
            schema=StructType([
                StructField("text", StringType()),
                StructField("language", StringType())
            ]),
            language=languages.get(lang_code.value, "Unknown")
        )

    def __init__(self, result):
        self._language = result.language

    def eval(self, row, lang_code: str):
        # Tag each row with the pre-resolved language name
        yield (row['text'], self._language)
$$;

SELECT * FROM tag_language(
  TABLE(VALUES ('Hola mundo'), ('Buenos días') t(text)),
  'es'
);

Output

+-------------+----------+
| text        | language |
+-------------+----------+
| Hola mundo  | Spanish  |
| Buenos días | Spanish  |
+-------------+----------+

Pour plus de modèles et de détails sur le transfert d'état, consultez Transférer l'état aux futurs appels eval.

Spécifier le partitionnement à partir de la méthode analyze

Lorsqu'une UDTF polymorphe accepte un argument de table, la méthode analyze peut contrôler la manière dont les lignes d'entrée sont distribuées entre les instances UDTF en définissant partitionBy, orderBy, withSinglePartition et select sur le AnalyzeResult. Ceci élimine la nécessité pour les appelants de spécifier PARTITION BY ou ORDER BY en SQL.

Pour l'API de partitionnement complète et des exemples, consultez Spécifier un partitionnement des lignes d'entrée de la méthode analyze.

Isolation de l'environnement

remarque

Les environnements d'isolation partagés nécessitent Databricks Runtime 17.2 ou une version ultérieure. Dans les versions antérieures, toutes les UDTF Python d'Unity Catalog s'exécutent en mode d'isolation strict.

Les UDTF Python d'Unity Catalog ayant le même propriétaire et la même session peuvent partager un environnement d'isolation par default. Cela améliore les performances et réduit la consommation de mémoire en réduisant le nombre d'environnements séparés qui doivent être lancés.

Isolement strict

Pour garantir qu'une UDTF s'exécute toujours dans son propre environnement, entièrement isolé, ajoutez la clause caractéristique STRICT ISOLATION.

La plupart des UDTF n'ont pas besoin d'un isolement strict. Les UDTF de traitement de données standard bénéficient de l'environnement d'isolation partagé default et s'exécutent plus rapidement avec une consommation de mémoire inférieure.

Ajoutez la clause caractéristique STRICT ISOLATION aux UDTF qui :

• Exécutez l'entrée en tant que code à l'aide de eval(), exec() ou de fonctions similaires.
• Écrire des fichiers dans le système de fichiers local.
• Modifier les variables globales ou l'état du système.
• Accéder ou modifier les variables d'environnement.

L'exemple UDTF suivant définit une variable d'environnement personnalisée, relit la variable et multiplie un ensemble de nombres à l'aide de la variable. Étant donné que l'UDTF modifie l'environnement du processus, exécutez-le dans STRICT ISOLATION. Autrement, cela pourrait divulguer ou remplacer les variables d'environnement d'autres UDF/UDTF dans le même environnement, provoquant un comportement incorrect.

SQL

CREATE OR REPLACE TEMPORARY FUNCTION multiply_numbers(factor STRING)
RETURNS TABLE (original INT, scaled INT)
LANGUAGE PYTHON
STRICT ISOLATION
HANDLER 'Multiplier'
AS $$
import os

class Multiplier:
    def eval(self, factor: str):
        # Save the factor as an environment variable
        os.environ["FACTOR"] = factor

        # Read it back and convert it to a number
        scale = int(os.getenv("FACTOR", "1"))

        # Multiply 0 through 4 by the factor
        for i in range(5):
            yield (i, i * scale)
$$;

SELECT * FROM multiply_numbers("3");

Définissez DETERMINISTIC si votre fonction produit des résultats cohérents

Ajoutez DETERMINISTIC à la définition de votre fonction si elle produit les mêmes sorties pour les mêmes entrées. Cela permet aux optimisations de query d'améliorer les performances.

By default, les UDTF Python Batch Unity Catalog sont supposées être non déterministes, sauf si elles sont explicitement déclarées. Les exemples de fonctions non déterministes incluent : générer des valeurs aléatoires, accéder aux heures ou aux dates actuelles, ou effectuer des appels d’API externes.

Voir CREATE FUNCTION (SQL, Python, Scala et Java).

Exemples pratiques

Les exemples suivants illustrent des cas d'usage réels pour les UDTF Python du Unity Catalog, allant de simples Transformations de données à des intégrations externes complexes.

Exemple : Réimplémentation explode

Alors que Spark fournit une fonction explode intégrée, la création de votre propre version démontre le schéma UDTF fondamental consistant à prendre une seule entrée et à produire plusieurs lignes de sortie.

SQL

CREATE OR REPLACE FUNCTION my_explode(arr ARRAY<STRING>)
RETURNS TABLE (element STRING)
LANGUAGE PYTHON
HANDLER 'MyExplode'
DETERMINISTIC
AS $$
class MyExplode:
    def eval(self, arr):
        if arr is None:
            return
        for element in arr:
            yield (element,)
$$;

Utilisez la fonction directement dans une query SQL :

SQL

SELECT element FROM my_explode(array('apple', 'banana', 'cherry'));

Output

+---------+
| element |
+---------+
| apple   |
| banana  |
| cherry  |
+---------+

Ou appliquez-le aux données de table existantes avec une LATERAL jointure:

SQL

SELECT s.*, e.element
FROM my_items AS s,
LATERAL my_explode(s.items) AS e;

Exemple : géolocalisation d'adresse IP via l'API REST

Cet exemple montre comment les UDTFs peuvent intégrer des APIs externes directement dans votre workflow SQL. Les analystes peuvent enrichir des données avec des appels d'API en temps réel en utilisant une syntaxe SQL familière, sans nécessiter de processus ETL distincts.

SQL

CREATE OR REPLACE FUNCTION ip_to_location(ip_address STRING)
RETURNS TABLE (city STRING, country STRING)
LANGUAGE PYTHON
HANDLER 'IPToLocationAPI'
AS $$
class IPToLocationAPI:
    def eval(self, ip_address):
        import requests
        api_url = f"https://api.ip-lookup.example.com/{ip_address}"
        try:
            response = requests.get(api_url)
            response.raise_for_status()
            data = response.json()
            yield (data.get('city'), data.get('country'))
        except requests.exceptions.RequestException as e:
            # Return nothing if the API request fails
            return
$$;

remarque

Les UDTF Python autorisent le trafic réseau TCP/UDP sur les ports 80, 443 et 53 lorsque vous utilisez un compute serverless ou un compute configuré avec le mode d'accès standard.

Utilisez la fonction pour enrichir les données de logs web avec des informations géographiques :

SQL

SELECT
  l.timestamp,
  l.request_path,
  geo.city,
  geo.country
FROM web_logs AS l,
LATERAL ip_to_location(l.ip_address) AS geo;

Cette approche permet l'analyse géographique en temps réel sans nécessiter de tables de recherche prétraitées ou de pipelines de données distincts. L'UDTF gère les requêtes HTTP, l'analyse JSON et la gestion des erreurs, rendant les sources de données externes accessibles via des requêtes SQL standard.

Exemple : faites correspondre les adresses IP à des blocs réseau CIDR.

Cet exemple illustre la correspondance des adresses IP avec les blocs réseau CIDR, une tâche d'ingénierie des données courante qui nécessite une logique SQL complexe.

Tout d'abord, créez des exemples de données avec des adresses IPv4 et IPv6 :

SQL

-- An example IP logs with both IPv4 and IPv6 addresses
CREATE OR REPLACE TEMPORARY VIEW ip_logs AS
VALUES
  ('log1', '192.168.1.100'),
  ('log2', '10.0.0.5'),
  ('log3', '172.16.0.10'),
  ('log4', '8.8.8.8'),
  ('log5', '2001:db8::1'),
  ('log6', '2001:db8:85a3::8a2e:370:7334'),
  ('log7', 'fe80::1'),
  ('log8', '::1'),
  ('log9', '2001:db8:1234:5678::1')
t(log_id, ip_address);

Ensuite, définissez et enregistrez l'UDTF. Notez la structure de classe Python :

• Le parameter t TABLE accepte une table d'entrée avec n'importe quel schéma. L'UDTF s'adapte automatiquement pour traiter toutes les colonnes fournies. Cette flexibilité signifie que vous pouvez utiliser la même fonction sur différentes tables sans modifier la signature de la fonction. Cependant, vous devez vérifier attentivement le schéma des lignes pour assurer la compatibilité.
• La méthode __init__ est utilisée pour une configuration lourde et unique, comme le chargement de la grande liste de réseaux. Ce travail a lieu une fois par partition de la table d'entrée.
• La méthode eval traite chaque ligne et contient la logique de correspondance principale. Cette méthode s'exécute exactement une fois pour chaque ligne de la partition d'entrée, et chaque exécution est effectuée par l'instance correspondante de la classe UDTF IpMatcher pour cette partition.
• La clause HANDLER spécifie le nom de la classe Python qui implémente la logique UDTF.

SQL

CREATE OR REPLACE TEMPORARY FUNCTION ip_cidr_matcher(t TABLE)
RETURNS TABLE(log_id STRING, ip_address STRING, network STRING, ip_version INT)
LANGUAGE PYTHON
HANDLER 'IpMatcher'
COMMENT 'Match IP addresses against a list of network CIDR blocks'
AS $$
class IpMatcher:
    def __init__(self):
        import ipaddress
        # Heavy initialization - load networks once per partition
        self.nets = []
        cidrs = ['192.168.0.0/16', '10.0.0.0/8', '172.16.0.0/12',
                 '2001:db8::/32', 'fe80::/10', '::1/128']
        for cidr in cidrs:
            self.nets.append(ipaddress.ip_network(cidr))

    def eval(self, row):
        import ipaddress
	    # Validate that required fields exist
        required_fields = ['log_id', 'ip_address']
        for field in required_fields:
            if field not in row:
                raise ValueError(f"Missing required field: {field}")
        try:
            ip = ipaddress.ip_address(row['ip_address'])
            for net in self.nets:
                if ip in net:
                    yield (row['log_id'], row['ip_address'], str(net), ip.version)
                    return
            yield (row['log_id'], row['ip_address'], None, ip.version)
        except ValueError:
            yield (row['log_id'], row['ip_address'], 'Invalid', None)
$$;

Maintenant que ip_cidr_matcher est enregistré dans Unity Catalog, appelez-le directement depuis SQL en utilisant la syntaxe TABLE() :

SQL

-- Process all IP addresses
SELECT
  *
FROM
  ip_cidr_matcher(t => TABLE(ip_logs))
ORDER BY
  log_id;

Output

+--------+-------------------------------+-----------------+-------------+
| log_id | ip_address                    | network         | ip_version  |
+--------+-------------------------------+-----------------+-------------+
| log1   | 192.168.1.100                 | 192.168.0.0/16  | 4           |
| log2   | 10.0.0.5                      | 10.0.0.0/8      | 4           |
| log3   | 172.16.0.10                   | 172.16.0.0/12   | 4           |
| log4   | 8.8.8.8                       | null            | 4           |
| log5   | 2001:db8::1                   | 2001:db8::/32   | 6           |
| log6   | 2001:db8:85a3::8a2e:370:7334  | 2001:db8::/32   | 6           |
| log7   | fe80::1                       | fe80::/10       | 6           |
| log8   | ::1                           | ::1/128         | 6           |
| log9   | 2001:db8:1234:5678::1         | 2001:db8::/32   | 6           |
+--------+-------------------------------+-----------------+-------------+

Exemple : Légendage d'images par batch à l'aide des Endpoint de vision Databricks

Cet exemple illustre l'annotation d'images par batch à l'aide d'un endpoint de service de modèle de vision Databricks. Elle présente l'utilisation de terminate() pour le traitement par batch et l'exécution basée sur des partitions.

1. Créer une table avec des URLs d'image publiques :

   SQL

   CREATE OR REPLACE TEMPORARY VIEW sample_images AS
   VALUES
       ('https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg', 'scenery'),
       ('https://upload.wikimedia.org/wikipedia/commons/thumb/a/a7/Camponotus_flavomarginatus_ant.jpg/1024px-Camponotus_flavomarginatus_ant.jpg', 'animals'),
       ('https://upload.wikimedia.org/wikipedia/commons/thumb/1/15/Cat_August_2010-4.jpg/1200px-Cat_August_2010-4.jpg', 'animals'),
       ('https://upload.wikimedia.org/wikipedia/commons/thumb/c/c5/M101_hires_STScI-PRC2006-10a.jpg/1024px-M101_hires_STScI-PRC2006-10a.jpg', 'scenery')
   images(image_url, category);

2. Créer une UDTF Python Unity Catalog pour générer des légendes d'image :

   1. Initialisez l'UDTF avec la configuration, y compris la taille de batch, le jeton d'API Databricks, l'Endpoint du modèle de vision et l'URL de l'espace de travail.
   2. Dans la méthode eval, recueillez les URL d'images dans un tampon. Lorsque le tampon atteint la taille de batch, Trigger le traitement par batchs. Cela garantit que plusieurs images sont traitées ensemble dans un seul appel d'API plutôt que des appels individuels par image.
   3. Dans la méthode de traitement par batch, download toutes les images mises en mémoire tampon, encodez-les en base64, et envoyez-les dans une seule requête API à Databricks VisionModel. Le modèle traite toutes les images simultanément et renvoie des légendes pour l'ensemble du batch.
   4. La méthode terminate est exécutée exactement une fois à la fin de chaque partition. Dans la méthode de terminaison, traitez les images restantes dans le tampon et restituez toutes les légendes collectées sous forme de résultats.

remarque

Remplacez <workspace-url> par l'URL de votre Databricks Workspace (https://your-workspace.cloud.databricks.com).

SQL

CREATE OR REPLACE TEMPORARY FUNCTION batch_inference_image_caption(data TABLE, api_token STRING)
RETURNS TABLE (caption STRING)
LANGUAGE PYTHON
HANDLER 'BatchInferenceImageCaption'
COMMENT 'batch image captioning by sending groups of image URLs to a Databricks vision endpoint and returning concise captions for each image.'
AS $$
class BatchInferenceImageCaption:
    def __init__(self):
        self.batch_size = 3
        self.vision_endpoint = "databricks-claude-sonnet-4-5"
        self.workspace_url = "<workspace-url>"
        self.image_buffer = []
        self.results = []

    def eval(self, row, api_token):
        self.image_buffer.append((str(row[0]), api_token))
        if len(self.image_buffer) >= self.batch_size:
            self._process_batch()

    def terminate(self):
        if self.image_buffer:
            self._process_batch()
        for caption in self.results:
            yield (caption,)

    def _process_batch(self):
        batch_data = self.image_buffer.copy()
        self.image_buffer.clear()

        import base64
        import httpx
        import requests

        # API request timeout in seconds
        api_timeout = 60
        # Maximum tokens for vision model response
        max_response_tokens = 300
        # Temperature controls randomness (lower = more deterministic)
        model_temperature = 0.3

        # create a batch for the images
        batch_images = []
        api_token = batch_data[0][1] if batch_data else None

        for image_url, _ in batch_data:
            image_response = httpx.get(image_url, timeout=15)
            image_data = base64.standard_b64encode(image_response.content).decode("utf-8")
            batch_images.append(image_data)

        content_items = [{
            "type": "text",
            "text": "Provide brief captions for these images, one per line."
        }]
        for img_data in batch_images:
            content_items.append({
                "type": "image_url",
                "image_url": {
                    "url": "data:image/jpeg;base64," + img_data
                }
            })

        payload = {
            "messages": [{
                "role": "user",
                "content": content_items
            }],
            "max_tokens": max_response_tokens,
            "temperature": model_temperature
        }

        response = requests.post(
            self.workspace_url + "/serving-endpoints/" +
            self.vision_endpoint + "/invocations",
            headers={
                'Authorization': 'Bearer ' + api_token,
                'Content-Type': 'application/json'
            },
            json=payload,
            timeout=api_timeout
        )

        result = response.json()
        batch_response = result['choices'][0]['message']['content'].strip()

        lines = batch_response.split('\n')
        captions = [line.strip() for line in lines if line.strip()]

        while len(captions) < len(batch_data):
            captions.append(batch_response)

        self.results.extend(captions[:len(batch_data)])
$$;

Pour utiliser l'UDTF de légende d'image par batch, appelez-la à l'aide de la table d'exemples d'images :

remarque

Remplacez your_secret_scope et api_token par le Secret Scope et le nom de clé réels pour le jeton d'API Databricks.

SQL

SELECT
  caption
FROM
  batch_inference_image_caption(
    data => TABLE(sample_images),
    api_token => secret('your_secret_scope', 'api_token')
  )

Output

+---------------------------------------------------------------------------------------------------------------+
| caption                                                                                                       |
+---------------------------------------------------------------------------------------------------------------+
| Wooden boardwalk cutting through vibrant wetland grasses under blue skies                                     |
| Black ant in detailed macro photography standing on a textured surface                                        |
| Tabby cat lounging comfortably on a white ledge against a white wall                                          |
| Stunning spiral galaxy with bright central core and sweeping blue-white arms against the black void of space. |
+---------------------------------------------------------------------------------------------------------------+

Vous pouvez également générer des légendes d'image catégorie par catégorie :

SQL

SELECT
  *
FROM
  batch_inference_image_caption(
    TABLE(sample_images)
    PARTITION BY category ORDER BY (category),
    secret('your_secret_scope', 'api_token')
  )

Output

+------------------------------------------------------------------------------------------------------+
| caption                                                                                              |
+------------------------------------------------------------------------------------------------------+
| Black ant in detailed macro photography standing on a textured surface                               |
| Stunning spiral galaxy with bright center and sweeping blue-tinged arms against the black of space.  |
| Tabby cat lounging comfortably on white ledge against white wall                                     |
| Wooden boardwalk cutting through lush wetland grasses under blue skies                               |
+------------------------------------------------------------------------------------------------------+

Exemple : calcul des courbes ROC et de l’AUC pour l’évaluation des modèles ML

Cet exemple montre comment calculer les courbes de caractéristiques de fonctionnement du récepteur (ROC) et les scores de l'aire sous la courbe (AUC) pour l'évaluation du modèle de classification binaire à l'aide de scikit-learn.

Cet exemple présente plusieurs modèles importants :

• **Utilisation de bibliothèques externes** : intègre scikit-learn pour le calcul des courbes ROC.
• Agrégation avec état : accumule les prédictions sur toutes les lignes avant de calculer les métriques.
• Utilisation de la méthodeterminate() : Traite l'ensemble du dataset et ne produit des résultats qu'après que toutes les lignes ont été évaluées.
• Gestion des erreurs : valide que les colonnes requises existent dans la table d’entrée

L'UDTF accumule toutes les prédictions en mémoire à l'aide de la méthode eval(), puis compute et génère la courbe ROC complète dans la méthode terminate(). Ce modèle est utile pour les métriques qui nécessitent le dataset complet pour le calcul.

SQL

CREATE OR REPLACE TEMPORARY FUNCTION compute_roc_curve(t TABLE)
RETURNS TABLE (threshold DOUBLE, true_positive_rate DOUBLE, false_positive_rate DOUBLE, auc DOUBLE)
LANGUAGE PYTHON
HANDLER 'ROCCalculator'
COMMENT 'Compute ROC curve and AUC using scikit-learn'
AS $$
class ROCCalculator:
    def __init__(self):
        from sklearn import metrics
        self._roc_curve = metrics.roc_curve
        self._roc_auc_score = metrics.roc_auc_score

        self._true_labels = []
        self._predicted_scores = []

    def eval(self, row):
        if 'y_true' not in row or 'y_score' not in row:
            raise KeyError("Required columns 'y_true' and 'y_score' not found")

        true_label = row['y_true']
        predicted_score = row['y_score']

        label = float(true_label)
        self._true_labels.append(label)
        self._predicted_scores.append(float(predicted_score))

    def terminate(self):
        false_pos_rate, true_pos_rate, thresholds = self._roc_curve(
            self._true_labels,
            self._predicted_scores,
            drop_intermediate=False
        )

        auc_score = float(self._roc_auc_score(self._true_labels, self._predicted_scores))

        for threshold, tpr, fpr in zip(thresholds, true_pos_rate, false_pos_rate):
            yield float(threshold), float(tpr), float(fpr), auc_score
$$;

Créer des données d'exemple de classification binaire avec des prédictions :

SQL

CREATE OR REPLACE TEMPORARY VIEW binary_classification_data AS
SELECT *
FROM VALUES
  ( 1, 1.0, 0.95, 'high_confidence_positive'),
  ( 2, 1.0, 0.87, 'high_confidence_positive'),
  ( 3, 1.0, 0.82, 'medium_confidence_positive'),
  ( 4, 0.0, 0.78, 'false_positive'),
  ( 5, 1.0, 0.71, 'medium_confidence_positive'),
  ( 6, 0.0, 0.65, 'false_positive'),
  ( 7, 0.0, 0.58, 'true_negative'),
  ( 8, 1.0, 0.52, 'low_confidence_positive'),
  ( 9, 0.0, 0.45, 'true_negative'),
  (10, 0.0, 0.38, 'true_negative'),
  (11, 1.0, 0.31, 'low_confidence_positive'),
  (12, 0.0, 0.15, 'true_negative'),
  (13, 0.0, 0.08, 'high_confidence_negative'),
  (14, 0.0, 0.03, 'high_confidence_negative')
AS data(sample_id, y_true, y_score, prediction_type);

Calculer la courbe ROC et l’AUC :

SQL

SELECT
    threshold,
    true_positive_rate,
    false_positive_rate,
    auc
FROM compute_roc_curve(
  TABLE(
    SELECT y_true, y_score
    FROM binary_classification_data
    WHERE y_true IS NOT NULL AND y_score IS NOT NULL
    ORDER BY sample_id
  )
)
ORDER BY threshold DESC;

Output

+-----------+---------------------+----------------------+-------+
| threshold | true_positive_rate  | false_positive_rate  | auc   |
+-----------+---------------------+----------------------+-------+
| 1.95      | 0.0                 | 0.0                  | 0.786 |
| 0.95      | 0.167               | 0.0                  | 0.786 |
| 0.87      | 0.333               | 0.0                  | 0.786 |
| 0.82      | 0.5                 | 0.0                  | 0.786 |
| 0.78      | 0.5                 | 0.125                | 0.786 |
| 0.71      | 0.667               | 0.125                | 0.786 |
| 0.65      | 0.667               | 0.25                 | 0.786 |
| 0.58      | 0.667               | 0.375                | 0.786 |
| 0.52      | 0.833               | 0.375                | 0.786 |
| 0.45      | 0.833               | 0.5                  | 0.786 |
| 0.38      | 0.833               | 0.625                | 0.786 |
| 0.31      | 1.0                 | 0.625                | 0.786 |
| 0.15      | 1.0                 | 0.75                 | 0.786 |
| 0.08      | 1.0                 | 0.875                | 0.786 |
| 0.03      | 1.0                 | 1.0                  | 0.786 |
+-----------+---------------------+----------------------+-------+

Exemple : projection dynamique de colonne à partir d'un argument de table

Cet exemple combine des fonctions UDTF polymorphes avec des arguments de table. L'UDTF accepte une table et une liste de noms de colonnes séparées par des virgules, puis projette uniquement ces colonnes de l'entrée. La méthode analyze inspecte le schéma de la table d'entrée et construit un schéma de sortie contenant uniquement les colonnes demandées.

SQL

CREATE OR REPLACE FUNCTION project_columns(t TABLE, columns STRING)
RETURNS TABLE
LANGUAGE PYTHON
HANDLER 'ProjectColumns'
AS $$
class ProjectColumns:
    @staticmethod
    def analyze(t, columns):
        from pyspark.sql.types import StructType
        from pyspark.sql.udtf import AnalyzeResult

        requested = [c.strip() for c in columns.value.split(",")]
        input_schema = t.dataType
        output_fields = []
        for field in input_schema.fields:
            if field.name in requested:
                output_fields.append(field)
        if not output_fields:
            raise ValueError(
                f"None of the requested columns {requested} "
                f"exist in the input table"
            )
        return AnalyzeResult(schema=StructType(output_fields))

    def eval(self, row, columns: str):
        requested = [c.strip() for c in columns.split(",")]
        yield tuple(row[col] for col in requested if col in row)
$$;

Utilisez la fonction pour sélectionner des colonnes spécifiques d'une table :

SQL

SELECT * FROM project_columns(
  TABLE(SELECT * FROM samples.nyctaxi.trips LIMIT 5),
  'pickup_zip, dropoff_zip, fare_amount'
);

Limitations

Les limitations suivantes s'appliquent aux UDTF Python de Unity Catalog :

• Les identifiants de service Unity Catalog ne sont pas pris en charge.
• Les dépendances personnalisées ne sont pas prises en charge.

Ressources supplémentaires

• UDTF Python délimitées par la session
• Aperçu des UDF régies par Unity Catalog