注意:以下翻译的准确性尚未经过验证。这是使用 AIP ↗ 从原始英文文本进行的机器翻译。
Python 函数目前处于测试阶段,可能无法在所有注册中使用。
以下是当前支持的函数 API 类型的完整列表,它们对应的 Python 类型,以及是否可以使用其对应的 Python 类型而不是函数 API 类型进行声明:
函数 API 类型 | 可以声明为 Python 类型吗? | 对应的 Python 类型 |
---|---|---|
Array | 是 | list |
Binary | 是 | bytes |
Boolean | 是 | bool |
Byte | 否 | int* |
Date | 是 | datetime.date |
Decimal | 是 | decimal.Decimal |
Double | 否 | float* |
Float | 是 | float |
Integer | 是 | int |
Long | 否 | int* |
Map | 是 | dict |
Set | 是 | set |
Short | 否 | int* |
字符串 | 是 | str |
Timestamp | 是 | datetime.datetime |
结构体/自定义类型 | 是 | dataclass.dataclass (示例) |
虽然 Integer
和 Long
都对应于 Python 类型 int
,但在函数签名中直接标记为 int
的字段将注册为 Integer
类型。因此,我们建议使用 API 中的 Integer
或 Long
类型来注册数值数据类型。类似的指导适用于 Float
和 Double
;如果 Python 类型 float
直接在您的函数签名中,则默认将注册为 Float
。
下面显示了另一个带有输入的函数示例:
Copied!1 2 3 4 5 6 7
from functions.api import function, Long, String, Timestamp @function def get_end_day_of_week(start_time: Timestamp, elapsed_millis: Long) -> String: # 这里是函数逻辑。这个函数的目的是根据给定的开始时间(Timestamp)和经过的毫秒数(Long), # 计算并返回一周的结束日期(String)。 pass
如上面的类型表所示,此函数也可以仅使用内置Python类型声明:
Copied!1 2 3 4 5 6 7
from functions.api import function from datetime import datetime @function def get_end_day_of_week(start_time: datetime, elapsed_millis: int) -> str: # 这里是函数逻辑 pass
在这个代码片段中,我们定义了一个装饰函数 @function
,并创建了一个函数 get_end_day_of_week
,它接受两个参数:start_time
(类型为 datetime
)和 elapsed_millis
(类型为整数,表示经过的毫秒数)。该函数应该返回一个字符串类型的结果,表示一周中的结束日期。函数的具体实现逻辑尚未编写。
此外,您可以结合使用内置类型和API类型:
Copied!1 2 3 4 5 6 7
from functions.api import function, Long, String from datetime import datetime @function def get_end_day_of_week(start_time: datetime, elapsed_millis: Long) -> String: # 在这里编写函数逻辑 pass
由其他支持类型组成的自定义Python类也可以被用于在函数签名中。API包没有明确的Custom
类型;相反,自定义类型被声明为用户定义的Python类。
要成为有效的自定义类型,类必须符合以下要求:
__init__
方法必须仅接受与字段具有相同名称和类型注解的命名参数。dataclasses.dataclass
↗装饰器可用于自动生成符合这些要求的__init__
方法。
dataclasses.dataclass
的自定义类型示例以下是使用dataclasses.dataclass
的示例:
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
from dataclasses import dataclass from functions.api import Float, Integer, String, function @dataclass class InventoryItem: """Class for keeping track of an item in inventory. 用于记录库存中商品的类。 """ name: String # 商品名称 unit_price: Float # 单价 quantity_on_hand: Integer = 0 # 库存数量,默认为0 def total_cost(self) -> Float: # 计算库存商品的总成本 return self.unit_price * self.quantity_on_hand @function def custom_type_with_init_from_decorator(inventory_item: InventoryItem) -> Float: # 使用装饰器定义的函数,接收一个InventoryItem对象并返回其总成本 return inventory_item.total_cost()
dataclass.dataclass
)以下是不使用dataclass.dataclass
而使用API类型的示例:
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
from functions.api import function, String, Timestamp class Event: timestamp: Timestamp # 时间戳属性,类型为Timestamp message: String # 消息属性,类型为String def __init__( self, timestamp: Timestamp, message: String ): self.timestamp = timestamp # 初始化时间戳 self.message = message # 初始化消息 @function def get_event_message(event: Event) -> String: return event.message # 返回事件的消息属性
此代码定义了一个Event
类,用于表示带有时间戳和消息的事件。get_event_message
函数用于从Event
对象中提取和返回消息。
了解如何设置Python函数库,或了解更多关于在Pipeline搭建器或Workshop中使用Python函数的信息。