用 Python 编写 UDTF¶

定义 process 方法¶

定义 process 方法，该方法接收从 SQL 类型转换的 UDTF 实参作为值，返回 Snowflake 将用于创建 UDTF 的表格返回值的数据。

方法的签名必须采用以下形式：

def process(self, *args):

process 方法必须满足以下条件：

• 有 self 参数。

• 声明与 UDTF 参数对应的方法参数。

  方法参数名称不必与 UDTF 参数名称匹配，但方法参数的声明顺序必须与 UDTF 参数的声明 顺序相同。

  将 UDTF 实参值传递到方法时，Snowflake 会将值从 SQL 类型转换为您在方法中使用的 Python 类型。有关 Snowflake 如何在 SQL 和 Python 数据类型之间映射的信息，请参阅 SQL-Python 数据类型映射。

• 生成一个或多个元组（或返回包含元组的可迭代对象），其中元组的序列对应 UDTF 返回值列的序列。

  元组元素的显示顺序必须与 UDTF 返回值列的声明 顺序相同。有关更多信息，请参阅 返回值 （本主题内容）。

  Snowflake 会将值从 Python 类型转换为 UDTF 声明所需的 SQL 类型。有关 Snowflake 如何在 SQL 和 Python 数据类型之间映射的信息，请参阅 SQL-Python 数据类型映射。

如果处理程序类中的方法引发异常，则处理将停止。调用 UDTF 的查询将失败，并显示错误消息。如果 process 方法返回 None，则处理将停止。（即使 process 方法返回 None，仍会调用 end_partition 方法。）

process 方法示例

以下示例中的代码显示 StockSale 处理程序类，该类具有 process 方法，用于处理三个 UDTF 实参（symbol、quantity 和 price），返回单个行，其中包含两列（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 子句中返回值声明的列的顺序。

  以下示例中的代码返回表示两行的值。

  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。

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

  Copy

跳过行¶

要跳过输入行并处理下一行（例如，在验证输入行时），请让 process 方法返回以下其中一项：

• 使用 return 时，返回 None、包含 None 的列表或空列表，以跳过该行。

• 使用 yield 时，返回 None，以跳过一行。

  请注意，如果多次调用 yield，则 Snowflake 会忽略返回 None 的调用后的任何调用。

以下示例中的代码仅返回其 number 为正整数的行。如果 number 不是正数，则该方法返回 None，以跳过当前行并继续处理下一行。

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

有状态处理和无状态处理¶

您可以实现处理程序以分区感知方式处理行，或者简单地逐行处理。

• 在 分区感知处理 中，处理程序包括用于管理分区范围状态的代码。这包括在分区处理开始时执行的 __init__ 方法，以及 Snowflake 在处理分区的最后一行后调用的 end_partition 方法。有关更多信息，请参阅 处理分区 （本主题内容）。

• 在 分区无感知处理 中，处理程序无状态执行，忽略分区边界。

  要以这种方式执行处理程序，请不要包含 __init__ 或 end_partition 方法。

完成分区处理¶

您可以选择在处理程序类中实现 end_partition 方法。Snowflake 在您处理分区中的所有行后调用该方法。通过此方法，您可以在处理完分区的所有行后执行分区的代码。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 子句指定的顺序（如果有）接收行数据。