Pythonでの UDTF の記述¶

process メソッドの定義¶

SQL型から変換されたUDTF引数を値として受け取る process メソッドを定義し、SnowflakeがUDTFの表形式の戻り値を作成するために使用するデータを返します。

メソッドのシグネチャは、次の形式である必要があります。

def process(self, *args):

process メソッドには次が必要です。

• self パラメータがある。

• UDTF パラメーターに対応するメソッドパラメーターを宣言する。

  メソッドパラメーター名はUDTFパラメータ名と一致する必要はありませんが、メソッドパラメーターはUDTFパラメータが宣言されるのと 同じ順序で 宣言される必要があります。

  UDTF引数値をメソッドに渡すと、Snowflakeは値をSQLタイプからメソッドで使用するPython型に変換します。SnowflakeがSQLとPythonのデータ型の間でどのようにマッピングされるかについては、 SQL-Pythonデータ型マッピング をご参照ください。

• タプルのシーケンスがUDTF戻り値列のシーケンスに対応する、1つ以上のタプルを生成します（またはタプルを含むイテレータを返します）。

  タプル要素は、UDTF戻り値列が宣言されているのと 同じ順序 で表示される必要があります。詳細については、このトピック内の 値を返す をご参照ください。

  Snowflakeは、値をPython型からUDTF宣言に必要なSQL型に変換します。SnowflakeがSQLとPythonのデータ型の間でどのようにマッピングされるかについては、 SQL-Pythonデータ型マッピング をご参照ください。

ハンドラークラスのメソッドが例外をスローすると、処理は停止します。UDTFを呼び出したクエリは、エラーメッセージを表示して失敗します。 process メソッドが None を返す場合、処理は停止します。（process メソッドが None を返しても、 end_partition メソッドは引き続き呼び出されます。）

processメソッドの例

次の例のコードは、3つのUDTF引数（symbol、 quantity、および price）を処理する process メソッドを持つ StockSale ハンドラークラスを示し、2つの列（symbol および total）を含む単一の行を返します。 process メソッドパラメーターは stock_sale パラメーターと同じ順序で宣言されていることに注意してください。 process メソッドの yield ステートメントの引数は、 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.8
handler='StockSale'
as $$
class StockSale:
    def process(self, symbol, quantity, price):
      cost = quantity * price
      yield (symbol, cost)
$$;

次の例のコードは、前述の UDF を呼び出し、 stocks_table テーブルの列 symbol、 quantity、および price から値を渡します。

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

値を返す¶

出力行を返す場合、 yield または return のいずれか（両方ではない）を使用して、表形式の値を持つタプルを返すことができます。メソッドが None を返すか生成する場合、現在の行の処理は停止します。

• yield を使用する場合は、出力行ごとに個別の yield ステートメントを実行します。 yield に付属する遅延評価により、より効率的な処理が可能になり、タイムアウトを回避できるため、これがベストプラクティスです。

  タプルの各要素は、UDTFによって返される結果の列値になります。 yield 引数の順序は、CREATE FUNCTIONのRETURNS TABLE句で戻り値として宣言された列の順序と一致する必要があります。

  次の例のコードは、2つの行を表す値を返します。

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

  Copy

  次の例のように、yield引数はタプルであるため、タプルに単一の値を渡す場合は、後続するにコンマを含める必要があることに注意してください。

  yield (cost,)
  

  Copy

• return を使用する場合、タプルで反復可能を返します。

  タプルの各値は、UDTFによって返される結果の列値になります。タプル内の列値の順序は、CREATE FUNCTIONのRETURNS TABLE句で戻り値として宣言された列の順序と一致する必要があります。

  次の例のコードは、symbolとtotalの2つの列を持つ2つの行を返します。

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

  Copy

行をスキップする¶

入力行をスキップして次の行を処理するには（入力行を検証する場合など）、 process メソッドに次のいずれかを返すようにします。

• return を使用する場合は、 None、 None を含むリスト、または行をスキップするための空のリストを返します。

• yield を使用する場合は、 None を返して行をスキップします。

  yield への呼び出しが複数ある場合、 None を返す呼び出しの後の呼び出しはSnowflakeによって無視されることに注意してください。

次の例のコードは、 number が正の整数である行のみを返します。 number が正ではない場合、メソッドは None を返し、現在の行をスキップして次の行の処理を続行します。

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

ステートフルおよびステートレス処理¶

ハンドラーを実装して、パーティション対応の方法で行を処理することも、行ごとに行を処理することもできます。

• パーティション対応処理 では、ハンドラーにパーティションスコープの状態を管理するためのコードが含まれます。これには、パーティション処理の開始時に実行される __init__ メソッドと、パーティションの最後の行の処理後にSnowflakeが呼び出す end_partition メソッドが含まれます。詳細については、このトピック内の パーティションの処理 をご参照ください。

• パーティションを認識しない処理 では、ハンドラーはパーティションの境界を無視してステートレスに実行されます。

  ハンドラーをこのように実行するには、 __init__ または end_partition メソッドを含めないでください。

パーティション処理の終了¶

オプションで、パーティション内のすべての行を処理した後にSnowflakeが呼び出す end_partition メソッドをハンドラークラスに実装できます。このメソッドでは、パーティションのすべての行が処理された後、パーティションのコードを実行できます。 end_partition メソッドは、パーティションスコープの計算結果を返すなどの出力行を生成する場合があります。詳細については、このトピック内の パーティションの処理 をご参照ください。

メソッドのシグネチャは、次の形式である必要があります。

def end_partition(self):

Snowflakeは、次の end_partition メソッドの実装を想定しています。

• 静的であってはなりません。

• self 以外のパラメータは含まれていない可能性があります。

• 表形式の値を返す代わりに、空のリストまたは None を生成する場合があります。

パーティション処理の例¶

次の例のコードは、（process メソッドで）最初に購入あたりのコストを計算し、購入を合計することによって、株式の購入全体で支払われる合計コストを計算します。このコードは、 end_partition メソッドで合計を返します。

UDTF の呼び出しと共に、このハンドラーを含んでいる UDTF の例については、 ハンドラークラスの例 をご参照ください。

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)

パーティションを処理するときは、次の点に注意してください。

• コードは、UDTFの呼び出しで明示的に指定されていないパーティションを処理する場合があります。UDTFの呼び出しにPARTITION BY句が含まれていない場合でも、Snowflakeはデータを暗黙的にパーティション化します。

• process メソッドは、パーティションのORDER BY句で指定された順序で行データを受け取ります（存在する場合）。