- 数据工程
- 大数据
- 序列化
- 数据分析
【免费下载链接】arrow
Apache Arrow is a multi-language toolbox for accelerated data interchange and in-memory processing
Apache Arrow 的pyarrow.compute模块为 Python 开发者提供了跨类型、向量化的计算函数库，覆盖聚合、累积、算术、位运算、字符串、时间、选择、排序与结构化变换等数百个函数。本文基于docs/source/python/compute.rst与docs/source/python/api/compute.rst，系统梳理 Compute Functions 的调用范式、分组聚合、Table/Dataset Join、表达式过滤及用户自定义函数（UDF）的完整实践，并结合仓库源码说明其实现机制。
一、模块速览与 API 参考定位
pyarrow.compute（简称pc）是 PyArrow 中所有标准计算操作的统一入口。官方 API 参考位于 docs/source/python/api/compute.rst，按功能类别组织为：
- 聚合（Aggregations）：
sum、mean、min、max、count、count_distinct、quantile、stddev、variance、approximate_median、tdigest、mode、product、min_max、index、all、any等 - 累积函数（Cumulative）：
cumulative_sum、cumulative_prod、cumulative_max、cumulative_min及其_checked变体 - 算术（Arithmetic）：
add、subtract、multiply、divide、power、abs、negate、sign、sqrt等，多数带_checked溢出检测变体 - 位运算（Bit-wise）：
bit_wise_and、bit_wise_or、bit_wise_xor、bit_wise_not、shift_left、shift_right - 取整（Rounding）：
ceil、floor、trunc、round、round_to_multiple - 对数（Logarithmic）：
ln、log2、log10、log1p、logb及其_checked变体 - 三角函数（Trigonometric）：
sin、cos、tan、asin、acos、atan、atan2等 - 比较（Comparisons）：
equal、not_equal、greater、greater_equal、less、less_equal、max_element_wise、min_element_wise - 逻辑（Logical）：
and_、or_、xor、invert、and_not，以及 Kleene 三值逻辑变体and_kleene、or_kleene、and_not_kleene - 字符串谓词/变换/填充/修剪/拆分/拼接/切片/包含测试：如
utf8_is_alnum、ascii_is_alpha、replace_substring、utf8_lower、ascii_lpad、split_pattern、extract_regex、binary_join、starts_with、match_substring等 - 分类（Categorizations）：
is_null、is_valid、is_nan、is_inf、is_finite、true_unless_null - 选择/多路复用（Selecting）：
if_else、coalesce、case_when、choose - 转换（Conversions）：
cast、strftime、strptime、ceil_temporal、floor_temporal、round_temporal、run_end_encode、run_end_decode - 时间分量提取与时间差：
year、month、day、hour、iso_week、day_of_week、days_between、months_between等 - 关联变换、选择、排序与结构化变换：
unique、value_counts、dictionary_encode、take、filter、sort_indices、partition_nth_indices、select_k_unstable、fill_null、list_flatten、struct_field、make_struct等 - Compute Options：
SortOptions、CountOptions、CastOptions、QuantileOptions、RoundOptions、SplitPatternOptions等约 50 个选项类 - 用户自定义函数（UDF）：
register_scalar_function、UdfContext
二、标准计算函数的调用范式
pc中的标准计算函数支持数组（Array / ChunkedArray）与标量（Scalar）两种输入。从源码 python/pyarrow/compute.py 看，模块启动时会遍历 C++ 函数注册表，为每个函数动态生成 Python 包装器：
- 若第一个参数是
Expression，则调用被转发为Expression._call（用于构建惰性表达式）； - 否则直接调用底层 C++ 内核
func.call(args, options, memory_pool)； - 多数函数接受一个可选的
memory_pool参数，用于控制执行期间的内存分配。
官方文档 docs/source/python/compute.rst 给出的最简示例：
>>> import pyarrow as pa >>> import pyarrow.compute as pc >>> a = pa.array([1, 1, 2, 3]) >>> pc.sum(a) <pyarrow.Int64Scalar: 7>元素级比较与标量运算：
>>> 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>输入限制
文档明确指出：多数函数同时支持数组与标量输入，但少数函数强制要求数组。典型如sort_indices，其唯一输入必须是数组（或 Table）。完整函数清单以 docs/source/python/api/compute.rst 为准。
溢出与域错误的_checked变体
文档强调：默认情况下算术与累积函数不检测溢出。需要安全计算时，使用带_checked后缀的变体，溢出或域错误会抛出ArrowInvalid异常。例如：
- 算术：
add_checked、subtract_checked、multiply_checked、divide_checked、power_checked、abs_checked、negate_checked、sqrt_checked - 累积：
cumulative_sum_checked、cumulative_prod_checked - 对数：
ln_checked、log10_checked、log2_checked、log1p_checked、logb_checked - 三角函数：
sin_checked、cos_checked、tan_checked、asin_checked、acos_checked
三、表排序示例：sort_indices
计算函数不仅能做逐元素运算，还能完成更复杂的表级操作。文档给出按列排序表的示例：
>>> import pyarrow as pa >>> import pyarrow.compute as pc >>> 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 ]返回的UInt64Array是排序后的行号索引，可用pc.take(t, i)还原排序后的表。sort_keys支持多列与升降序组合，底层由SortOptions驱动（见 python/pyarrow/compute.py 中SortOptions的导入）。
四、分组聚合（Grouped Aggregations）
文档指出：分组聚合函数在模块级调用会直接抛异常，必须通过Table.group_by()的声明式 API 使用。
基本用法
>>> import pyarrow as pa >>> 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 values_sum: int64 keys: string ---- values_sum: [[3,7,5]] keys: [["a","b","c"]]这里的"sum"聚合实际对应 C++ 侧名为hash_sum的计算函数。从源码结构看，group_by返回一个TableGroupBy声明对象（见 python/pyarrow/table.pxi 中Table.group_by），随后调用aggregate()才真正执行哈希聚合。group_by还支持use_threads参数（默认True），文档与源码均提示：开启多线程时不保证输出的稳定排序。
多聚合同时执行
>>> t.group_by("keys").aggregate([ ... ("values", "sum"), ... ("keys", "count") ... ]) pyarrow.Table values_sum: int64 keys_count: int64 keys: string ---- values_sum: [[3,7,5]] keys_count: [[2,2,1]] keys: [["a","b","c"]]结果列名规则为<列名>_<聚合名>，即values_sum、keys_count。
聚合选项：CountOptions
每个聚合函数都可附带对应的 Options 对象。文档以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 values_count: int64 keys: string ---- values_count: [[3]] keys: [["a"]] >>> table_with_nulls.group_by(["keys"]).aggregate([ ... ("values", "count", pc.CountOptions(mode="only_valid")) ... ]) pyarrow.Table values_count: int64 keys: string ---- values_count: [[1]] keys: [["a"]]mode="all"计入 null，mode="only_valid"只计非 null 值。所有支持的哈希聚合函数均可带"hash_"前缀或不带前缀使用（如"sum"与"hash_sum"等价）。完整清单由 C++ 侧hash_aggregate内核注册生成，可参阅 cpp/src/arrow/compute/registry.cc 及 cpp/src/arrow/compute/api_aggregate.h。
五、Table 与 Dataset 的 Join 操作
Table和Dataset均支持通过join()方法进行连接。文档明确了支持的 join 类型：
- left semi（左半连接）
- right semi（右半连接）
- left anti（左反连接）
- right anti（右反连接）
- inner（内连接）
- left outer（左外连接，默认）
- right outer（右外连接）
- full outer（全外连接）
基本 Join
import pyarrow as pa 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")结果（左外连接，未匹配行以 null 填充）：
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]]指定 join 类型与多键
table1.join(table2, keys='id', join_type="full outer")结果中未匹配的右侧行也会出现（id=4）：
pyarrow.Table id: int64 year: int64 n_legs: int64 animal: string ---- id: [[3,1,2,4]] year: [[2019,2020,2022,null]] n_legs: [[5,null,null,100]] animal: [["Brittle stars",null,null,"Centipede"]]多键连接：先给table2增加year列，再按("id", "year")双键连接：
table2_withyear = table2.append_column("year", pa.array([2019, 2022])) table1.join(table2_withyear, keys=["id", "year"])结果只有id=3, year=2019的行有匹配数据，其余为 null。
Dataset 的 Join
同样的能力对Dataset可用：
import pyarrow.dataset as ds ds1 = ds.dataset(table1) ds2 = ds.dataset(table2) joined_ds = ds1.join(ds2, keys="id")结果是一个InMemoryDataset，可通过joined_ds.head(5)预览。
从 python/pyarrow/table.pxi 的Table.join签名可以看到更多工程参数：right_keys（右侧键列名，默认与左侧同名）、left_suffix/right_suffix（列名冲突时加后缀）、coalesce_keys（是否从一侧合并去重键列，默认True）、use_threads（多线程，默认True）。
六、表达式过滤（Filtering by Expressions）
Table与Dataset都可以使用布尔型Expression进行过滤。表达式从pc.field()构建，field支持列名、整数下标以及嵌套字段引用（如pc.field(("a", "b"))），其定义见 python/pyarrow/compute.py 中field与scalar函数。
构建过滤表达式
文档示例：找出"nums"列中的偶数行。
import pyarrow.compute as pc even_filter = (pc.bit_wise_and(pc.field("nums"), pc.scalar(1)) == pc.scalar(0))原理：num & 1只保留最低位。二进制下末尾为1的数是奇数，num & 1结果为非零；末尾为0的数是偶数，结果为0。因此bit_wise_and(nums, 1) == 0精确选出所有偶数。
应用到表过滤：
>>> 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"]]组合与取反
多个过滤表达式可用&、|、~分别表示逻辑与、或、非：
>>> table.filter(~even_filter) # 奇数 pyarrow.Table nums: [[1,3,5,7,9]] chars: [["a","c","e","g","i"]] >>> table.filter(even_filter & (pc.field("nums") > 5)) # 偶数且大于5 pyarrow.Table nums: [[6,8,10]] chars: [["f","h","l"]]Dataset 的惰性过滤
Dataset.filter返回一个新的Dataset，过滤操作是惰性的——只有真正访问数据时才执行：
>>> 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"]]多个filter可以链式叠加，最终to_table()触发实际计算。
七、用户自定义函数（UDF）
⚠️ 该 API 为实验性功能（python/pyarrow/_compute.pyx 中
register_scalar_function亦明确标注 EXPERIMENTAL）。
PyArrow 允许注册自定义标量函数，注册后不仅能从 Python 调用，也能从 C++（以及任何包装 Arrow C++ 的语言，如 R 的arrow包）按函数名调用。标量函数执行逐元素运算，输出不依赖输入顺序，与 SQL 表达式中的函数或 NumPy 的 universal functions 大致对应。
注册一个 UDF
需要提供：函数名、函数文档（summary + description）、输入类型与输出类型。完整示例（实现最大公约数numpy_gcd）：
import numpy as np import pyarrow as pa import pyarrow.compute as pc 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)关键点（结合源码 python/pyarrow/_compute.pyx）：
- 函数实现第一个参数必须是context（示例中的
ctx），类型为pyarrow.compute.UdfContext； UdfContext暴露memory_pool属性，UDF 内部的内存分配应使用它；in_types是Dict[str, DataType]，参数名用于生成文档，参数个数决定函数元数（arity）；若函数定义为*args，最后一个in_type会被视为所有可变参数的公共类型；- 当所有参数都是标量时 UDF 应返回
Scalar，否则返回与输入等长的Array； - 可通过可选的
func_registry参数指定注册到非默认的函数注册表。
直接调用 UDF
使用pc.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 ]标量 × 标量返回Scalar，标量 × 数组返回Array，体现逐元素语义。
在 Dataset 投影表达式中使用 UDF
UDF 可用于任何按名字引用计算函数的场景，例如在 Dataset 投影中使用Expression._call。以下示例用numpy_gcd计算每行value列与 30 的最大公约数：
>>> import pyarrow.dataset as ds >>> 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，func_args中传入的是表达式而非立即求值的值（注意区分pyarrow.scalar构造标量值 与pyarrow.compute.scalar构造表达式）。该表达式由投影算子在实际执行时求值。
投影表达式的限制
"投影"是指为表动态增加新列。投影函数必须满足：对每一输入行输出恰好一个值，且该值只由当前行计算得出，不依赖其他行。因此：
- ✅
numpy_gcd这类逐行纯函数可用于投影； - ❌
cumulative_sum不行——每个结果依赖前面所有行； - ❌
drop_null/filter不行——并非每行都有输出。
八、Compute Options 概览
许多计算函数通过 Options 类定制行为。API 参考中列出的选项类包括：ArraySortOptions、AssumeTimezoneOptions、CastOptions、CountOptions、CumulativeSumOptions、DayOfWeekOptions、DictionaryEncodeOptions、ElementWiseAggregateOptions、ExtractRegexOptions、FilterOptions、IndexOptions、JoinOptions、ListSliceOptions、MakeStructOptions、MapLookupOptions、MatchSubstringOptions、ModeOptions、NullOptions、PadOptions、PairwiseOptions、PartitionNthOptions、QuantileOptions、ReplaceSliceOptions、ReplaceSubstringOptions、RoundOptions、RoundTemporalOptions、RoundToMultipleOptions、RunEndEncodeOptions、ScalarAggregateOptions、SelectKOptions、SetLookupOptions、SliceOptions、SortOptions、SplitOptions、SplitPatternOptions、StrftimeOptions、StrptimeOptions、StructFieldOptions、TakeOptions、TDigestOptions、TrimOptions、VarianceOptions、WeekOptions等。
从 python/pyarrow/compute.py 源码可以看到，这些选项类全部从 C++ 侧pyarrow._compute导入，且 Python 层在生成每个函数的签名时，会内省对应 Options 类的构造参数并拼接到函数签名中——因此实际使用时可写作pc.quantile(a, q=0.5)或pc.quantile(a, options=QuantileOptions(q=0.5))，两种方式等价。同时每个函数还暴露memory_pool关键字参数，用于指定执行内存池。
九、实现机制：从 Python 包装到 C++ 内核
pc模块的动态生成机制值得展开说明（python/pyarrow/compute.py）：
- 模块导入时调用
function_registry()获取全局 C++ 函数注册表； - 遍历
list_functions()，对每个函数调用_wrap_function生成包装器； - 包装器通过
__arrow_compute_function__属性携带函数名、元数（arity）、选项类等元数据； - Python 关键字
and、or与 C++ 函数名冲突，自动改写为and_、or_； hash_aggregate类型与零元标量聚合（如count_all）不会暴露为模块级可调用函数——这正是分组聚合必须走Table.group_by的原因；- 若首个参数是
Expression，包装器返回Expression._call(...)构建惰性计算图，否则直接执行 C++ 内核func.call(...)。
C++ 侧的注册与内核实现在 cpp/src/arrow/compute/registry.cc（包含RegisterAggregateOptions等注册逻辑）与 cpp/src/arrow/compute/api_aggregate.h（聚合内核 API）。因此pc中数百个函数实际上都是对同一套 C++ 计算内核的 Python 绑定。
十、结语
pyarrow.compute是 PyArrow 数据处理的"瑞士军刀"：单函数覆盖标量与数组、模块级自动包装 C++ 内核、_checked变体提供溢出/域错误防护、Table.group_by承接哈希聚合、Table/Dataset.join提供八种连接类型、Expression支撑惰性过滤与投影、UDF 打通 Python 与 C++ 的函数边界。结合 docs/source/python/api/compute.rst 的完整函数清单与 docs/source/python/compute.rst 的实战示例，即可在数据管道中高效落地这些能力。
- 数据工程
- 大数据
- 序列化
- 数据分析
【免费下载链接】arrow
Apache Arrow is a multi-language toolbox for accelerated data interchange and in-memory processing
相关推荐
PyArrow Compute Functions 完全指南：聚合、Join、表达式过滤与 UDF 实战
PyArrow Compute Functions 完全指南：聚合、Join、表达式过滤与 UDF 实战 <output_article PyArrow Com
数据工程数据分析大数据Polars 聚合实战指南：基于 `group_by` 的表达式聚合、分组内过滤与排序
Polars 聚合实战指南：基于 group_by 的表达式聚合、分组内过滤与排序 本文以 Polars 官方用户指南《Aggregation》为主体，围绕 g
数据分析大数据Apache Arrow 计算函数完全指南：分组聚合、连接、表达式过滤与用户自定义函数
Apache Arrow 计算函数完全指南：分组聚合、连接、表达式过滤与用户自定义函数 导读 pyarrow.compute 是 Apache Arrow 提供
数据工程大数据序列化数据分析
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
