Écriture d’une UDTF en Python¶

Définition d’une méthode process¶

Définissez une méthode process qui reçoit comme valeurs les UDTF arguments convertis à partir de types SQL, renvoyant des données que Snowflake utilisera pour créer la valeur de retour tabulaire de l’UDTF.

La signature de la méthode doit être de la forme suivante :

def process(self, *args):

Votre méthode process doit :

• Avoir un paramètre self.

• Déclarer les paramètres de la méthode correspondant aux paramètres de l’UDTF.

  Les noms des paramètres des méthodes ne doivent pas nécessairement correspondre aux noms des paramètres de l’UDTF, mais les paramètres des méthodes doivent être déclarés dans le même ordre que les paramètres de l’UDTF.

  Lorsque vous transmettez des valeurs d’arguments de l’UDTF à votre méthode, Snowflake convertit les valeurs des types SQL en types Python que vous utilisez dans la méthode. Pour plus d’informations sur la manière dont Snowflake établit une correspondance entre SQL et les types de données Python, voir Mappages des types de données SQL-Python.

• Donne un ou plusieurs tuples (ou renvoie un itérable contenant des tuples), dans lequel la séquence de tuples correspond à la séquence de colonnes de valeurs de retour de l’UDTF.

  Les éléments du tuple doivent apparaître dans le même ordre que les colonnes de la valeur de retour UDTF sont déclarées. Pour plus d’informations, voir Renvoi d’une valeur dans cette rubrique.

  Snowflake convertira les valeurs des types Python en types SQL requis par la déclaration UDTF. Pour plus d’informations sur la manière dont Snowflake établit une correspondance entre SQL et les types de données Python, voir Mappages des types de données SQL-Python.

Si une méthode de la classe du gestionnaire lève une exception, le traitement s’arrête. La requête qui a appelé l’UDTF échoue avec un message d’erreur. Si la méthode process renvoie None, le traitement s’arrête. (La méthode end_partition est toujours invoquée même si la méthode process renvoie None).

Exemple de méthode de traitement

L’exemple suivant présente une classe de gestionnaire StockSale avec une méthode process qui traite trois arguments UDTF (symbol, quantity, et price), et renvoie une seule ligne avec deux colonnes (symbol et total). Notez que les paramètres de la méthode process sont déclarés dans le même ordre que les paramètres stock_sale. Les arguments de l’instruction yield de la méthode process sont dans le même ordre que les colonnes déclarées dans la clause stock_sale RETURNS TABLE.

CREATE OR REPLACE FUNCTION stock_sale(symbol VARCHAR, quantity NUMBER, price NUMBER(10,2))
  RETURNS TABLE (symbol VARCHAR, total NUMBER(10,2))
  LANGUAGE PYTHON
  RUNTIME_VERSION = 3.9
  HANDLER = 'StockSale'
AS $$
class StockSale:
    def process(self, symbol, quantity, price):
      cost = quantity * price
      yield (symbol, cost)
$$;

Le code de l’exemple suivant appelle l’UDF précédente, en transmettant les valeurs des colonnes symbol, quantity, et price de la table stocks_table.

SELECT stock_sale.symbol, total
  FROM stocks_table, TABLE(stock_sale(symbol, quantity, price) OVER (PARTITION BY symbol));

Renvoi d’une valeur¶

Lorsque vous renvoyez des lignes de sortie, vous pouvez utiliser soit yield soit return (mais pas les deux) pour renvoyer des tuples avec la valeur tabulaire. Si la méthode renvoie ou produit None, le traitement de la ligne en cours s’arrête.

• Lorsque vous utilisez yield, exécutez une instruction yield distincte pour chaque ligne de sortie. Il s’agit de la meilleure pratique, car l’évaluation paresseuse qui accompagne yield permet un traitement plus efficace et peut contribuer à éviter les dépassements de délai d’expiration.

  Chaque élément du tuple devient une valeur de colonne dans le résultat renvoyé par l’UDTF. L’ordre des arguments yield doit correspondre à l’ordre des colonnes déclarées pour la valeur de retour dans la clause RETURNS TABLE de CREATE FUNCTION.

  Le code de l’exemple suivant renvoie des valeurs représentant deux rangées.

  def process(self, symbol, quantity, price):
    cost = quantity * price
    yield (symbol, cost)
    yield (symbol, cost)
  

  Copy

  Notez que, comme l’argument de rendement est un tuple, vous devez inclure une virgule de fin lorsque vous passez une seule valeur dans le tuple, comme dans l’exemple suivant.

  yield (cost,)
  

  Copy

• En utilisant return, renvoyez un itérable avec des tuples.

  Chaque valeur d’un tuple devient une valeur de colonne dans le résultat renvoyé par l’UDTF. L’ordre des valeurs des colonnes dans un tuple doit correspondre à l’ordre des colonnes déclarées pour la valeur de retour dans la clause RETURNS TABLE de CREATE FUNCTION.

  Le code de l’exemple suivant renvoie deux lignes, chacune avec deux colonnes : symbole et total.

  def process(self, symbol, quantity, price):
    cost = quantity * price
    return [(symbol, cost), (symbol, cost)]
  

  Copy

Ignorer des lignes¶

Pour ignorer une ligne d’entrée et traiter la ligne suivante (par exemple, lorsque vous validez les lignes d’entrée), faites en sorte que la méthode process renvoie l’un des éléments suivants :

• Lorsque vous utilisez return, vous devez retourner None, une liste contenant None, ou une liste vide pour ignorer la ligne.

• Lorsque vous utilisez yield, retournez None pour ignorer une ligne.

  Notez que si vous avez plusieurs appels à yield, tout appel après un appel qui renvoie None sera ignoré par Snowflake.

Le code de l’exemple suivant renvoie uniquement les lignes pour lesquelles number est un nombre entier positif. Si number n’est pas positif, la méthode renvoie None vide pour ignorer la ligne actuelle et poursuivre le traitement de la ligne suivante.

def process(self, number):
  if number < 1:
    yield None
  else:
    yield (number)

Traitement avec et sans état¶

Vous pouvez implémenter le gestionnaire pour traiter les lignes en tenant compte des partitions ou pour les traiter simplement ligne par ligne.

• Dans le traitement en tenant compte des partitions, le gestionnaire inclut du code pour gérer l’état spécifique à la partition. Cela comprend une méthode __init__ qui s’exécute au début du traitement de la partition et une méthode end_partition que Snowflake invoque après avoir traité la dernière ligne de la partition. Pour plus d’informations, voir Traitement des partitions dans cette rubrique.

• Dans le traitement tenant compte de la partition, le gestionnaire s’exécute sans état, en ignorant les limites des partitions.

  Pour que le gestionnaire s’exécute de cette façon, il ne faut pas inclure de méthode __init__ ou end_partition.

Finalisation du traitement des partitions¶

Vous pouvez éventuellement implémenter une méthode end_partition dans votre classe de gestionnaire que Snowflake invoquera après avoir traité toutes les lignes d’une partition. Avec cette méthode, vous pouvez exécuter le code pour une partition après que toutes les lignes de la partition aient été traitées. Votre méthode end_partition peut produire des lignes de sortie, par exemple pour renvoyer les résultats d’un calcul à l’échelle de la partition. Pour plus d’informations, voir Traitement des partitions dans cette rubrique.

La signature de la méthode doit être de la forme suivante :

def end_partition(self):

Snowflake attend de l’implémentation de votre méthode end_partition ce qui suit :

• Elle ne doit pas être statique.

• Elle ne peut avoir d’autres paramètres que self.

• Au lieu de renvoyer une valeur tabulaire, elle peut produire une liste vide ou None.

Exemple de traitement des partitions¶

Le code de l’exemple suivant calcule le coût total payé sur l’ensemble des achats pour une action en calculant d’abord le coût par achat et en additionnant les achats (dans la méthode process). Le code renvoie le total dans la méthode end_partition.

Pour un exemple d’une UDTF qui inclut ce gestionnaire, ainsi que l’appel de l’UDTF, reportez-vous à Exemple de classe de gestionnaire.

class StockSaleSum:
  def __init__(self):
    self._cost_total = 0
    self._symbol = ""

  def process(self, symbol, quantity, price):
    self._symbol = symbol
    cost = quantity * price
    self._cost_total += cost
    yield (symbol, cost)

  def end_partition(self):
    yield (self._symbol, self._cost_total)

Lors du traitement des partitions, gardez à l’esprit les points suivants :

• Votre code peut gérer des partitions qui ne sont pas explicitement spécifiées dans un appel à l’UDTF. Même lorsqu’un appel à l’UDTF n’inclut pas de clause PARTITION BY, Snowflake partitionne les données de manière implicite.

• Votre méthode process recevra les données des lignes dans l’ordre spécifié par la clause ORDER BY de la partition, le cas échéant.