企业官网建设流程全解析

PyArrow Compute Functions 完全指南：逐元素运算、分组聚合、连接、表达式过滤与自定义 UDF

【免费下载链接】arrowApache Arrow is the universal columnar format and multi-language toolbox for fast data interchange and in-memory analytics项目地址: https://gitcode.com/GitHub_Trending/arrow3/arrow

本文以 Apache Arrow 官方 Python 文档 compute.rst 为主体，系统讲解pyarrow.compute模块的完整能力版图：从直接调用计算函数（pc.sum、pc.equal等）、通过Table.group_by做分组聚合，到Table/Dataset的 join 操作、基于表达式（Expression）的行过滤、实验性的用户自定义函数（UDF）注册，以及数组/标量上的标准 Python 运算符重载。读完本文，你将能够针对数组、标量、表乃至数据集选择合适的 compute 入口，并能通过函数注册机制把 NumPy 等生态的算法接入 Arrow 计算体系。

1. 计算模块的总体结构：函数注册表驱动的统一 API

Arrow 支持对"可能具有不同类型"的输入执行逻辑计算操作（logical compute operations）。标准计算操作由pyarrow.compute模块提供，可以直接调用：

>>> import pyarrow as pa >>> import pyarrow.compute as pc >>> a = pa.array([1, 1, 2, 3]) >>> pc.sum(a) <pyarrow.Int64Scalar: 7>

从源码结构看，pyarrow.compute中暴露的每个顶层函数并不是逐一手写的。compute.py 中的_make_global_functions()会遍历 C++ 侧的全局FunctionRegistry（function_registry()），为每个已注册的 compute 函数动态生成一个带签名、带文档字符串的 Python 包装器；hash_aggregate类函数因为不可直接调用（见第 3 节）而被显式跳过。每个包装器的文档由_decorate_compute_function从 C++ 侧的函数元数据（func._doc中的 summary、description、参数名、选项类）拼装而成，选项参数还能以关键字参数或options=整体两种形式传入（参见 compute.py 中的_handle_options）。

这套动态生成的机制对应着 C++ 侧的统一 Compute API：函数存储在arrow::compute::FunctionRegistry中按名称查找，计算输入统一表示为Datum（Scalar、Array、ChunkedArray 等形状的标签联合）。C++ 侧的函数注册与内核（kernel）实现位于 cpp/src/arrow/compute/，其中 registry.cc 管理注册表，kernels/目录存放各函数的具体实现，api_scalar.cc、api_aggregate.cc 等提供具体 API。完整函数清单见 C++ 计算文档中的 Available functions 一节（Python 文档中的.. arrow-computefuncs::指令即从该注册表自动生成函数列表）。

一个实用提示：大多数 compute 函数同时支持数组（含 chunked）与标量输入，但部分函数强制要求特定输入形状。例如sort_indices要求其第一个（也是唯一的）输入必须是数组。

2. 标准计算函数：逐元素运算、多值返回与表级操作

以下示例展示了数组间与标量间的基本调用方式：

>>> a = pa.array([1, 1, 2, 3]) >>> b = pa.array([4, 1, 2, 8]) >>> pc.equal(a, b) <pyarrow.lib.BooleanArray object at ...> [ false, true, true, false ] >>> x, y = pa.scalar(7.8), pa.scalar(9.3) >>> pc.multiply(x, y) <pyarrow.DoubleScalar: 72.54>

2.1 多值返回：StructScalar

如果一个 compute 函数返回多个值，结果会以StructScalar形式给出。可以通过调用其values()方法提取各个字段：

>>> a = pa.array([1, 1, 2, 3]) >>> pc.min_max(a) <pyarrow.StructScalar: [('min', 1), ('max', 3)]> >>> a, b = pc.min_max(a).values() >>> a <pyarrow.Int64Scalar: 1> >>> b <pyarrow.Int64Scalar: 3>

2.2 超越逐元素：对表进行排序

compute 函数不仅能做元素级运算，还能作用于整张表。例如按某一列排序并取回行索引：

>>> t = pa.table({'x':[1,2,3],'y':[3,2,1]}) >>> i = pc.sort_indices(t, sort_keys=[('y', 'ascending')]) >>> i <pyarrow.lib.UInt64Array object at ...> [ 2, 1, 0 ]

完整的 PyArrow 计算函数参考清单见官方 API 文档中的pyarrow.compute参考（compute.rst 末尾指向的api.compute参考页），跨语言的 C++ 函数全集可在 C++ Compute Functions 文档 中查阅。

3. 分组聚合：group_by 与 hash_* 聚合函数

分组聚合函数（grouped aggregation）不能像普通函数那样直接调用，而必须通过pyarrow.Table.group_by能力使用。group_by会返回一个分组声明（grouping declaration），在其上应用哈希聚合函数：

>>> t = pa.table([ ... pa.array(["a", "a", "b", "b", "c"]), ... pa.array([1, 2, 3, 4, 5]), ... ], names=["keys", "values"]) >>> t.group_by("keys").aggregate([("values", "sum")]) pyarrow.Table keys: string values_sum: int64 ---- keys: [["a","b","c"]] values_sum: [[3,7,5]]

上例中传给aggregate的"sum"聚合，底层就是hash_sumcompute 函数。group_by的 Python 实现见 table.pxi 中的Table.group_by方法。

3.1 一次执行多个聚合

>>> t = pa.table([ ... pa.array(["a", "a", "b", "b", "c"]), ... pa.array([1, 2, 3, 4, 5]), ... ], names=["keys", "values"]) >>> t.group_by("keys").aggregate([ ... ("values", "sum"), ... ("keys", "count") ... ]) pyarrow.Table keys: string values_sum: int64 keys_count: int64 ---- keys: [["a","b","c"]] values_sum: [[3,7,5]] keys_count: [[2,2,1]]

3.2 为聚合函数提供选项

每个聚合函数都可以提供选项。例如用CountOptions改变 null 值的计数方式：

>>> table_with_nulls = pa.table([ ... pa.array(["a", "a", "a"]), ... pa.array([1, None, None]) ... ], names=["keys", "values"]) >>> table_with_nulls.group_by(["keys"]).aggregate([ ... ("values", "count", pc.CountOptions(mode="all")) ... ]) pyarrow.Table keys: string values_count: int64 ---- keys: [["a"]] values_count: [[3]] >>> table_with_nulls.group_by(["keys"]).aggregate([ ... ("values", "count", pc.CountOptions(mode="only_valid")) ... ]) pyarrow.Table keys: string values_count: int64 ---- keys: [["a"]] values_count: [[1]]

CountOptions的mode对应三种计数语义：默认只统计非 null 值、只统计 null 值、或统计全部值。

3.3 受支持的分组聚合函数清单

所有支持的分组聚合函数都可以在aggregate中带或不带"hash_"前缀使用。其底层实现在 C++ 侧注册，例如 hash_aggregate.cc 中注册了hash_count、hash_count_all、hash_first、hash_last、hash_min_max、hash_min、hash_max、hash_any、hash_all、hash_count_distinct、hash_distinct、hash_one、hash_list等函数（数值类函数另见 hash_aggregate_numeric.cc）。结合 C++ 计算文档 的函数表，主要的分组聚合函数及其行为如下：

C++ 文档给出的一个典型例子值得注意：对包含 null 键与 null 值的输入做分组求和时，null 会被当作一个独立的键值参与分组（例如key=null的组单独成组，其sum(x)只由非 null 值累加得出）。这一点在编写涉及缺失分组键的 ETL 逻辑时需要留意。

4. Table 与 Dataset 的 Join 操作

pyarrow.Table与pyarrow.dataset.Dataset都通过各自的join方法支持连接操作（Python 侧实现见 table.pxi 中的Table.join）。方法接受要连接进来的右侧表/数据集，以及一个或多个连接键。默认执行left outer join，也可以请求以下任意连接类型：

• left semi
• right semi
• left anti
• right anti
• inner
• left outer
• right outer
• full outer

4.1 单键连接

只提供表和连接键即可完成基本连接：

>>> table1 = pa.table({'id': [1, 2, 3], ... 'year': [2020, 2022, 2019]}) >>> table2 = pa.table({'id': [3, 4], ... 'n_legs': [5, 100], ... 'animal': ["Brittle stars", "Centipede"]}) >>> joined_table = table1.join(table2, keys="id")

结果是一张由table1与table2在id键上执行 left outer join 得到的新表：

>>> joined_table pyarrow.Table id: int64 year: int64 n_legs: int64 animal: string ---- id: [[3,1,2]] year: [[2019,2020,2022]] n_legs: [[5,null,null]] animal: [["Brittle stars",null,null]]

4.2 指定连接类型：full outer join

通过join_type参数可以请求其他连接类型，例如全外连接。注意 join 结果中各列可能来自不同分块，示例中用combine_chunks()合并后再sort_by使输出可读：

>>> table1.join(table2, keys='id', join_type="full outer").combine_chunks().sort_by('id') pyarrow.Table id: int64 year: int64 n_legs: int64 animal: string ---- id: [[1,2,3,4]] year: [[2020,2022,2019,null]] n_legs: [[null,null,5,100]] animal: [[null,null,"Brittle stars","Centipede"]]

4.3 复合键连接

可以提供一个以上的连接键，使连接发生在两个键上。例如为table2增加year列后，按("id", "year")连接：

>>> table2_withyear = table2.append_column("year", pa.array([2019, 2022])) >>> table1.join(table2_withyear, keys=["id", "year"]) pyarrow.Table id: int64 year: int64 n_legs: int64 animal: string ---- id: [[3,1,2]] year: [[2019,2020,2022]] n_legs: [[5,null,null]] animal: [["Brittle stars",null,null]]

4.4 Dataset 级别的连接

Dataset.join具备同样的能力，可以直接把两个数据集连接起来：

>>> import pyarrow.dataset as ds >>> ds1 = ds.dataset(table1) >>> ds2 = ds.dataset(table2) >>> joined_ds = ds1.join(ds2, keys="id") >>> joined_ds.head(5) pyarrow.Table id: int64 year: int64 n_legs: int64 animal: string ---- id: [[3,1,2]] year: [[2019,2020,2022]] n_legs: [[5,null,null]] animal: [["Brittle stars",null,null]]

5. 表达式过滤：pc.field、布尔组合与惰性执行

Table与Dataset都可以用布尔类型的Expression进行过滤。表达式从pyarrow.compute.field构建起：对一个或多个字段施加比较与转换运算，即可组合出所需的过滤表达式。大多数 compute 函数都可以用于对field做转换。

pc.field与pc.scalar的 Python 实现分别见 compute.py。注意二者与pa.field/pa.scalar的本质区别：pyarrow.scalar()创建 Arrow 内存模型中的Scalar对象，而pyarrow.compute.scalar()创建的是代表标量值的Expression对象，用于计算表达式、谓词和数据集过滤。

5.1 用位运算构造"偶数"过滤器

下面构建一个找出列"nums"中所有偶数的过滤器：

>>> even_filter = (pc.bit_wise_and(pc.field("nums"), pc.scalar(1)) == pc.scalar(0))

其原理：1的二进制是00000001，只有末位为1的数（即奇数）与1做bit_wise_and才会得到非零结果，因此num & 1 == 0恰好刻画偶数。

构建好过滤器后，将其传给Table.filter即可只保留匹配行：

>>> table = pa.table({'nums': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], ... 'chars': ["a", "b", "c", "d", "e", "f", "g", "h", "i", "l"]}) >>> table.filter(even_filter) pyarrow.Table nums: int64 chars: string ---- nums: [[2,4,6,8,10]] chars: [["b","d","f","h","l"]]

5.2 组合过滤器：and / or / not

多个过滤器可以用&、|、~分别表达 and、or、not。例如~even_filter就是筛选所有奇数：

>>> table.filter(~even_filter) pyarrow.Table nums: int64 chars: string ---- nums: [[1,3,5,7,9]] chars: [["a","c","e","g","i"]]

也可以把even_filter与pc.field("nums") > 5组合，筛选出大于 5 的偶数：

>>> table.filter(even_filter & (pc.field("nums") > 5)) pyarrow.Table nums: int64 chars: string ---- nums: [[6,8,10]] chars: [["f","h","l"]]

5.3 Dataset 的惰性过滤

Dataset同样可以用Dataset.filter方法过滤。该方法返回一个新的Dataset实例，过滤器只有在真正访问数据时才被应用（惰性求值），因此多个filter调用可以链式组合而不产生中间物化：

>>> dataset = ds.dataset(table) >>> filtered = dataset.filter(pc.field("nums") < 5).filter(pc.field("nums") > 2) >>> filtered.to_table() pyarrow.Table nums: int64 chars: string ---- nums: [[3,4]] chars: [["c","d"]]

6. 用户自定义函数（UDF，实验性 API）

注意：官方文档明确标注该 API 为实验性（experimental）。

PyArrow 允许定义并注册自定义 compute 函数。注册后，这些函数可以从 Python、C++ 以及任何包装 Arrow C++ 的实现（如 R 的arrow包）按注册名调用。

UDF 支持范围限于标量函数（scalar function）：即对数组或标量执行逐元素操作的函数，其输出通常不依赖于参数中值的顺序。这类函数大致对应 SQL 表达式中使用的函数，或 NumPy 的 universal functions。

6.1 注册一个 UDF

注册 UDF 需要定义函数名、函数文档、输入类型和输出类型，使用pyarrow.compute.register_scalar_function：

>>> import numpy as np >>> function_name = "numpy_gcd" >>> function_docs = { ... "summary": "Calculates the greatest common divisor", ... "description": ... "Given 'x' and 'y' find the greatest number that divides\n" ... "evenly into both x and y." ... } >>> input_types = { ... "x" : pa.int64(), ... "y" : pa.int64() ... } >>> output_type = pa.int64() >>> >>> def to_np(val): ... if isinstance(val, pa.Scalar): ... return val.as_py() ... else: ... return np.array(val) >>> >>> def gcd_numpy(ctx, x, y): ... np_x = to_np(x) ... np_y = to_np(y) ... return pa.array(np.gcd(np_x, np_y)) >>> >>> pc.register_scalar_function(gcd_numpy, ... function_name, ... function_docs, ... input_types, ... output_type)

UDF 实现函数的第一个参数始终是context（上例中命名为ctx），它是pyarrow.compute.UdfContext的实例。该上下文暴露若干有用属性，特别是UdfContext.memory_pool，用于在 UDF 内部做内存分配（应使用 Arrow 内存池分配，而非普通 Python 对象默认分配）。

6.2 直接调用 UDF：call_function

>>> pc.call_function("numpy_gcd", [pa.scalar(27), pa.scalar(63)]) <pyarrow.Int64Scalar: 9> >>> pc.call_function("numpy_gcd", [pa.scalar(27), pa.array([81, 12, 5])]) <pyarrow.lib.Int64Array object at ...> [ 27, 3, 1 ]

可见 UDF 同样遵循标量/数组广播语义：(scalar, scalar)产出标量，(scalar, array)产出数组。

6.3 在 Dataset 中使用 UDF

更一般地，UDF 可以用在任何"可以按名称引用 compute 函数"的地方。例如通过Expression._call在数据集的列上调用它。考虑数据在表中，需要计算某一列与标量 30 的 GCD，复用上面注册的"numpy_gcd"：

>>> data_table = pa.table({'category': ['A', 'B', 'C', 'D'], 'value': [90, 630, 1827, 2709]}) >>> dataset = ds.dataset(data_table) >>> func_args = [pc.scalar(30), ds.field("value")] >>> dataset.to_table( ... columns={ ... 'gcd_value': ds.field('')._call("numpy_gcd", func_args), ... 'value': ds.field('value'), ... 'category': ds.field('category') ... }) pyarrow.Table gcd_value: int64 value: int64 category: string ---- gcd_value: [[30,30,3,3]] value: [[90,630,1827,2709]] category: [["A","B","C","D"]]

注意ds.field('')._call(...)返回的是一个pyarrow.compute.Expression。传给该函数调用的参数都是表达式而非标量值（注意pyarrow.scalar与pyarrow.compute.scalar的区别，后者产生表达式）。该表达式在投影算子（projection operator）执行时才被求值。

投影表达式的限制

在上例中我们用表达式为表添加了新列（gcd_value）。为表添加这种动态计算的新列称为投影（projection），对投影表达式中可以使用的函数有明确限制：

• 投影函数必须为每个输入行恰好输出一个值；
• 该输出值应完全由该行本身计算得出，不得依赖其他行。

因此上面一直使用的numpy_gcd是合法的投影函数；而"累积求和"（cumulative sum）不合法，因为它对某一行的结果依赖此前各行；"丢弃 null 行"（drop nulls）也不合法，因为它对某些行不产生输出值。

7. 标准 Python 运算符：数组与标量的运算符重载

PyArrow 为数组和标量支持标准 Python 运算符的逐元素运算。目前支持范围限于部分标准 compute 函数：算术运算（+、-、/、%、**）、位运算（&、|、^、>>、<<）及其他。

这些运算符尽可能使用底层内核的带检查（checked）版本，并带有相应的约束——例如两个字符串数组不能相加。使用示例：

>>> import pyarrow as pa >>> arr = pa.array([-1, 2, -3]) >>> val = pa.scalar(42.7) >>> arr + val <pyarrow.lib.DoubleArray object at ...> [ 41.7, 44.7, 39.7 ] >>> val ** arr <pyarrow.lib.DoubleArray object at ...> [ 0.023419203747072598, 1823.2900000000002, 0.000012844475506953143 ] >>> arr << 2 <pyarrow.lib.Int64Array object at ...> [ -4, 8, -12 ]

7.1 底层机制：隐式类型提升与 common numeric type

运算符重载最终落在 compute 内核上，其类型行为值得了解。从 C++ Compute 文档 看，当内核与参数类型不完全匹配时，函数可能先对参数做隐式转换：比较与算术内核要求同类型参数，支持通过把参数提升到"可容纳任一侧任意值的数值类型"来对不同数值类型求值，这一类型即common numeric type：

特别地，比较uint64列与int16列时，如果某个uint64值无法用公共类型int64表示（如2 ** 63），可能抛出错误。此外，算术函数默认版本不检测溢出（结果通常回绕），多数函数另有带"_checked"后缀的溢出检查变体，检测到溢出时返回Invalid错误状态——Python 侧运算符正是尽可能使用这些 checked 内核。

8. 小结：按场景选择计算入口

• 对数组/标量做逐元素运算或整体归约：直接用pyarrow.compute顶层函数（如pc.sum、pc.equal、pc.min_max），多值返回用StructScalar.values()解包；
• 按键分组汇总：使用Table.group_by(...).aggregate([...])，其中"sum"/"count"等聚合名对应hash_*compute 函数，并可通过CountOptions等选项类微调 null 语义；
• 表/数据集连接：Table.join/Dataset.join，默认 left outer，可指定 8 种连接类型与复合键；
• 行级过滤：用pc.field/pc.scalar构建Expression，配合& | ~组合，交给Table.filter（立即执行）或Dataset.filter（惰性执行）；
• 接入自有算法：通过register_scalar_function注册实验性 UDF，用pc.call_function直接调用，或在 Dataset 投影中以表达式形式按名称调用；
• 交互式脚本：对数值数组/标量可直接使用+、-、**、<<等 Python 运算符，底层映射到 checked 内核并遵循 common numeric type 提升规则。

以上行为均以当前仓库中 Python Compute 文档、C++ Compute 文档 与 pyarrow/compute.py 等源码为准；UDF 一节为实验性 API，生产使用前建议关注上游变更。

【免费下载链接】arrowApache Arrow is the universal columnar format and multi-language toolbox for fast data interchange and in-memory analytics项目地址: https://gitcode.com/GitHub_Trending/arrow3/arrow