Schreiben von UDTF in Python¶

Definieren einer process-Methode¶

Definieren Sie eine process-Methode, die als Werte die aus SQL-Typen konvertierten UDTF-Argumente erhält und Daten zurückgibt, die Snowflake zum Erstellen des tabellarischen Rückgabewerts der UDTF verwendet.

Die Signatur der Methode muss die folgende Form haben:

def process(self, *args):

Für Ihre process-Methode muss Folgendes gelten:

• Hat einen self-Parameter.

• Deklariert Methodenparameter, die den UDTF-Parametern entsprechen.

  Die Namen der Methodenparameter müssen nicht mit den Namen der UDTF-Parameter übereinstimmen, aber die Methodenparameter müssen in der gleichen Reihenfolge deklariert werden, wie die UDTF-Parameter deklariert sind.

  Wenn Sie UDTF-Argumentwerte an Ihre Methode übergeben, konvertiert Snowflake die SQL-Typen der Werte in die Python-Typen, die Sie in der Methode verwenden. Weitere Informationen zur Zuordnung von SQL-Datentypen zu Python-Datentypen in Snowflake finden Sie unter Zuordnung von Datentypen zwischen SQL und Python.

• Gibt ein oder mehrere Tupel aus (oder ein iterierendes Objekt mit Tupeln), wobei die Reihenfolge der Tupel der Reihenfolge der UDTF-Spalten für die Rückgabewerte entspricht.

  Die Tupel-Elemente müssen in der gleichen Reihenfolge erscheinen, wie die UDTF-Spalten für die Rückgabewerte deklariert sind. Weitere Informationen dazu finden Sie unter Zurückgeben eines Wertes (unter diesem Thema).

  Snowflake konvertiert dann die Python-Datentypen der Werte in die SQL-Datentypen, die von der UDTF-Deklaration benötigt werden. Weitere Informationen zur Zuordnung von SQL-Datentypen zu Python-Datentypen in Snowflake finden Sie unter Zuordnung von Datentypen zwischen SQL und Python.

Wenn eine Methode in der Handler-Klasse eine Ausnahme auslöst, wird die Verarbeitung angehalten. Die Abfrage, mit der die UDTF aufgerufen wurde, schlägt fehl und gibt eine Fehlermeldung aus. Wenn die process-Methode den Wert None zurückgibt, wird die Verarbeitung angehalten. (Die end_partition-Methode wird auch dann aufgerufen, wenn die process-Methode den Wert None zurückgibt).

Beispiel für „process“-Methode

Der Code im folgenden Beispiel zeigt eine Handler-Klasse StockSale mit einer process-Methode, die drei UDTF-Argumente (symbol, quantity und price) verarbeitet und eine einzelne Zeile mit zwei Spalten (symbol und total) zurückgibt. Beachten Sie, dass die Parameter der process-Methode in der gleichen Reihenfolge deklariert werden wie die stock_sale-Parameter. Die Argumente in der yield-Anweisung der process-Methode weisen dieselbe Reihenfolge auf wie die Spalten, die in der stock_sale RETURNS TABLE-Klausel deklariert sind.

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.8
handler='StockSale'
as $$
class StockSale:
    def process(self, symbol, quantity, price):
      cost = quantity * price
      yield (symbol, cost)
$$;

Im folgenden Beispiel wird die vorherige UDF aufgerufen, wobei Werte aus den Spalten symbol, quantity und price der Tabelle stocks_table übergeben werden.

select stock_sale.symbol, total
  from stocks_table, table(stock_sale(symbol, quantity, price) over (partition by symbol));

Zurückgeben eines Wertes¶

Bei der Rückgabe von Ausgabezeilen können Sie entweder yield oder return (aber nicht beide) verwenden, um Tupel mit dem Tabellenwert zurückzugeben. Wenn die Methode den Wert None zurückgibt oder ergibt, wird die Verarbeitung für die aktuelle Zeile angehalten.

• Wenn Sie yield verwenden, führen Sie für jede Ausgabezeile eine separate yield-Anweisung aus. Dies ist die beste Vorgehensweise, da die mit yield verbundene Auswertung im Lazy-Modus eine effizientere Verarbeitung ermöglicht und Timeouts vermeiden hilft.

  Jedes Element des Tupels wird zu einem Spaltenwert in dem von der UDTF zurückgegebenen Ergebnis. Die Reihenfolge der yield-Argumente muss mit der Reihenfolge der Spalten übereinstimmen, die in der RETURNS TABLE-Klausel der CREATE FUNCTION-Anweisung für den Rückgabewert deklariert sind.

  Der Code im folgenden Beispiel gibt Werte zurück, die zwei Zeilen darstellen.

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

  Copy

  Da das „yield“-Argument ein Tupel ist, müssen Sie bei Übergabe eines einzelnen Werts ein nachstehendes Komma einfügen, wie im folgenden Beispiel gezeigt:

  yield (cost,)
  

  Copy

• Bei Verwendung von return wird ein iterierendes Objekt mit Tupeln zurückgegeben.

  Jeder Wert in einem Tupel wird zu einem Spaltenwert in dem von der UDTF zurückgegebenen Ergebnis. Die Reihenfolge der Spaltenwerte in einem Tupel muss mit der Reihenfolge der Spalten übereinstimmen, die in der RETURNS TABLE-Klausel der CREATE FUNCTION-Anweisung für den Rückgabewert deklariert sind.

  Der Code im folgenden Beispiel gibt zwei Zeilen mit jeweils zwei Spalten, Symbol und Summe, zurück:

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

  Copy

Überspringen von Zeilen¶

Um eine Eingabezeile zu überspringen und die nächste Zeile zu verarbeiten (z. B. wenn die Eingabezeilen validiert werden), kann die process-Methode eine der folgenden Optionen zurückgeben:

• Bei Verwendung von return kann None, eine Liste mit None oder eine leere Liste zurückgegeben werden, um die Zeile zu überspringen.

• Bei Verwendung von yield kann None zurückgegeben werden, um eine Zeile zu überspringen.

  Beachten Sie, dass bei mehreren Aufrufen von yield alle Aufrufe nach einem Aufruf, der None zurückgibt, von Snowflake ignoriert werden.

Der Code im folgenden Beispiel gibt nur die Zeilen zurück, für die number eine positive ganze Zahl ist. Wenn number nicht positiv ist, gibt die Methode None zurück, um die aktuelle Zeile zu überspringen und mit der Verarbeitung der nächsten Zeile fortzufahren.

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

Zustandsabhängige und zustandslose Verarbeitung¶

Sie können bei der Implementierung des Handlers entscheiden, ob dieser Zeilen unter Beachtung von Partitionen verarbeitet oder einfach nur Zeile für Zeile verarbeitet.

• Bei der partitionsspezifischen Verarbeitung enthält der Handler Code zur Verwaltung des partitionsbezogenen Zustands. Dazu gehört eine __init__-Methode, die zu Beginn der Partitionsverarbeitung ausgeführt wird, und eine end_partition-Methode, die Snowflake nach der Verarbeitung der letzten Zeile der Partition aufruft. Weitere Informationen dazu finden Sie unter Verarbeiten von Partitionen (unter diesem Thema).

• Bei der partitionsunabhängigen Verarbeitung wird der Handler zustandslos ausgeführt und ignoriert Partitionsgrenzen.

  Damit der Handler auf diese Weise ausgeführt wird, dürfen Sie keine __init__- oder end_partition-Methode einschließen.

Abschließen der Partitionsverarbeitung¶

Sie können in Ihrer Handler-Klasse optional eine end_partition-Methode implementieren, die von Snowflake aufgerufen wird, nachdem alle Zeilen einer Partition verarbeitet wurden. Mit dieser Methode können Sie Code für eine Partition ausführen, nachdem alle Zeilen dieser Partition verarbeitet wurden. Ihre end_partition-Methode kann Ausgabezeilen erzeugen, um z. B. die Ergebnisse einer partitionsbezogenen Berechnung zurückzugeben. Weitere Informationen dazu finden Sie unter Verarbeiten von Partitionen (unter diesem Thema).

Die Signatur der Methode muss die folgende Form haben:

def end_partition(self):

Snowflake erwartet folgendes von der Implementierung Ihrer end_partition-Methode:

• Die Methode darf nicht statisch sein.

• Die Methode darf nur den Parameter self haben.

• Alternativ zur Rückgabe eines Tabellenwerts kann die Methode auch eine leere Liste oder None erzeugen.

Beispiel für die Verarbeitung von Partitionen¶

Der Code im folgenden Beispiel berechnet die Gesamtkosten, die für Käufe einer Aktie gezahlt wurden, indem zunächst die Kosten pro Kauf berechnet und die Käufe addiert werden (mit der Methode process). Der Code gibt die Summe in der Methode end_partition zurück.

Ein Beispiel für eine UDTF, die diesen Handler enthält, und den Aufruf der UDTF finden Sie unter Beispiel für eine Handler-Klasse.

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)

Bei der Verarbeitung von Partitionen ist Folgendes zu beachten:

• Ihr Code kann Partitionen verarbeiten, die nicht explizit in einem Aufruf der UDTF angegeben sind. Selbst wenn ein Aufruf der UDTF keine PARTITION BY-Klausel enthält, partitioniert Snowflake die Daten implizit.

• Ihre process-Methode empfängt die Zeilendaten in der Reihenfolge, die, falls vorhanden, in der ORDER BY-Klausel der Partition angegeben ist.