pyarrow.compute 完整函数指南：聚合、分组、Join 与表达式过滤实战

数据工程大数据序列化数据分析【免费下载链接】arrowApache Arrow is a multi-language toolbox for accelerated data interchange and in-memory processing项目地址https://gitcode.com/gh_mirrors/arrow13/arrow点击查看免费下载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按功能类别组织为聚合Aggregationssum、mean、min、max、count、count_distinct、quantile、stddev、variance、approximate_median、tdigest、mode、product、min_max、index、all、any等累积函数Cumulativecumulative_sum、cumulative_prod、cumulative_max、cumulative_min及其_checked变体算术Arithmeticadd、subtract、multiply、divide、power、abs、negate、sign、sqrt等多数带_checked溢出检测变体位运算Bit-wisebit_wise_and、bit_wise_or、bit_wise_xor、bit_wise_not、shift_left、shift_right取整Roundingceil、floor、trunc、round、round_to_multiple对数Logarithmicln、log2、log10、log1p、logb及其_checked变体三角函数Trigonometricsin、cos、tan、asin、acos、atan、atan2等比较Comparisonsequal、not_equal、greater、greater_equal、less、less_equal、max_element_wise、min_element_wise逻辑Logicaland_、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等分类Categorizationsis_null、is_valid、is_nan、is_inf、is_finite、true_unless_null选择/多路复用Selectingif_else、coalesce、case_when、choose转换Conversionscast、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 OptionsSortOptions、CountOptions、CastOptions、QuantileOptions、RoundOptions、SplitPatternOptions等约 50 个选项类用户自定义函数UDFregister_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(modeall)) ... ]) pyarrow.Table values_count: int64 keys: string ---- values_count: [[3]] keys: [[a]] table_with_nulls.group_by([keys]).aggregate([ ... (values, count, pc.CountOptions(modeonly_valid)) ... ]) pyarrow.Table values_count: int64 keys: string ---- values_count: [[1]] keys: [[a]]modeall计入 nullmodeonly_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全外连接基本 Joinimport 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, keysid)结果左外连接未匹配行以 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, keysid, join_typefull outer)结果中未匹配的右侧行也会出现id4pyarrow.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])结果只有id3, year2019的行有匹配数据其余为 null。Dataset 的 Join同样的能力对Dataset可用import pyarrow.dataset as ds ds1 ds.dataset(table1) ds2 ds.dataset(table2) joined_ds ds1.join(ds2, keysid)结果是一个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 ExpressionsTable与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_gcdimport 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.UdfContextUdfContext暴露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 投影表达式中使用 UDFUDF 可用于任何按名字引用计算函数的场景例如在 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.Expressionfunc_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, q0.5)或pc.quantile(a, optionsQuantileOptions(q0.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 的实战示例即可在数据管道中高效落地这些能力。赞分享数据工程大数据序列化数据分析【免费下载链接】arrowApache Arrow is a multi-language toolbox for accelerated data interchange and in-memory processing项目地址https://gitcode.com/gh_mirrors/arrow13/arrow点击查看免费下载相关推荐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 提供数据工程大数据序列化数据分析上一篇Open-Elevation自定义数据源如何导入和使用自定义高程数据集下一篇优化Basdonax AI RAG的Docker镜像构建多阶段构建创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考