如何在 Arrow C 上执行 compute 内核计算数组【免费下载链接】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当你在 C 里拿到一个 ArrowTable或Array想对它做求和、两列逐元素相加、在列中查找值这类计算时用到的就是 Arrow C 的 compute 模块。它把计算实现为按名字注册的 kernel你只需要初始化模块、准备输入数据、调用函数并检查返回的arrow::Status。本文基于官方教程 Arrow Compute 及其完整示例代码 compute_example.cc走通一条可以编译运行的路径覆盖三种典型调用方式带类型检查的初始化调用、便捷函数compute::Sum、通用入口compute::CallFunction以及带 Options 结构的index函数。前提是一个带 compute 模块的 Arrow 安装以及一段能产出Table的现有代码即对 Array、Table 等基础数据结构有基本认识这部分教程指向 basic_arrow 教程本文只用到其结论。准备条件构建带 compute 模块的 Arrow教程的前置要求见 compute_tutorial.rst一份 Arrow 安装构建方式见 build_system.rst如果你是自编译 Arrow必须启用 compute 模块即 CMake 参数-DARROW_COMPUTEON对应构建文档中的可选组件小节对 Arrow 基础数据结构Array、Table、Datum有基本了解。如果用的是系统发行版包或预编译包第 2 点通常已满足只有自编译时才需要显式确认该开关。完整示例代码下面是示例程序的完整代码与 compute_example.cc 一致先建一张两列 int32 的表然后依次做求和、逐元素相加、按值查下标三件事// (Doc section: Includes) #include arrow/api.h #include arrow/compute/api.h #include iostream // (Doc section: Includes) arrow::Status RunMain() { // (Doc section: Create Tables) // Create a couple 32-bit integer arrays. arrow::Int32Builder int32builder; int32_t some_nums_raw[5] {34, 624, 2223, 5654, 4356}; ARROW_RETURN_NOT_OK(int32builder.AppendValues(some_nums_raw, 5)); std::shared_ptrarrow::Array some_nums; ARROW_ASSIGN_OR_RAISE(some_nums, int32builder.Finish()); int32_t more_nums_raw[5] {75342, 23, 64, 17, 736}; ARROW_RETURN_NOT_OK(int32builder.AppendValues(more_nums_raw, 5)); std::shared_ptrarrow::Array more_nums; ARROW_ASSIGN_OR_RAISE(more_nums, int32builder.Finish()); // Make a table out of our pair of arrays. std::shared_ptrarrow::Field field_a, field_b; std::shared_ptrarrow::Schema schema; field_a arrow::field(A, arrow::int32()); field_b arrow::field(B, arrow::int32()); schema arrow::schema({field_a, field_b}); // Initialize the compute module to register the required compute kernels. ARROW_RETURN_NOT_OK(arrow::compute::Initialize()); std::shared_ptrarrow::Table table; table arrow::Table::Make(schema, {some_nums, more_nums}, 5); // (Doc section: Create Tables) // (Doc section: Sum Datum Declaration) // The Datum class is what all compute functions output to, and they can take Datums // as inputs, as well. arrow::Datum sum; // (Doc section: Sum Datum Declaration) // (Doc section: Sum Call) // Here, we can use arrow::compute::Sum. This is a convenience function, and the next // computation wont be so simple. However, using these where possible helps // readability. ARROW_ASSIGN_OR_RAISE(sum, arrow::compute::Sum({table-GetColumnByName(A)})); // (Doc section: Sum Call) // (Doc section: Sum Datum Type) // Get the kind of Datum and what it holds -- this is a Scalar, with int64. std::cout Datum kind: sum.ToString() content type: sum.type()-ToString() std::endl; // (Doc section: Sum Datum Type) // (Doc section: Sum Contents) // Note that we explicitly request a scalar -- the Datum cannot simply give what it is, // you must ask for the correct type. std::cout sum.scalar_asarrow::Int64Scalar().value std::endl; // (Doc section: Sum Contents) // (Doc section: Add Datum Declaration) arrow::Datum element_wise_sum; // (Doc section: Add Datum Declaration) // (Doc section: Add Call) // Get element-wise sum of both columns A and B in our Table. Note that here we use // CallFunction(), which takes the name of the function as the first argument. ARROW_ASSIGN_OR_RAISE(element_wise_sum, arrow::compute::CallFunction( add, {table-GetColumnByName(A), table-GetColumnByName(B)})); // (Doc section: Add Call) // (Doc section: Add Datum Type) // Get the kind of Datum and what it holds -- this is a ChunkedArray, with int32. std::cout Datum kind: element_wise_sum.ToString() content type: element_wise_sum.type()-ToString() std::endl; // (Doc section: Add Datum Type) // (Doc section: Add Contents) // This time, we get a ChunkedArray, not a scalar. std::cout element_wise_sum.chunked_array()-ToString() std::endl; // (Doc section: Add Contents) // (Doc section: Index Datum Declare) // Use an options struct to set up searching for 2223 in column A (the third item). arrow::Datum third_item; // (Doc section: Index Datum Declare) // (Doc section: IndexOptions Declare) // An options struct is used in lieu of passing an arbitrary amount of arguments. arrow::compute::IndexOptions index_options; // (Doc section: IndexOptions Declare) // (Doc section: IndexOptions Assign) // We need an Arrow Scalar, not a raw value. index_options.value arrow::MakeScalar(2223); // (Doc section: IndexOptions Assign) // (Doc section: Index Call) ARROW_ASSIGN_OR_RAISE( third_item, arrow::compute::CallFunction(index, {table-GetColumnByName(A)}, index_options)); // (Doc section: Index Call) // (Doc section: Index Inspection) // Get the kind of Datum and what it holds -- this is a Scalar, with int64 std::cout Datum kind: third_item.ToString() content type: third_item.type()-ToString() std::endl; // We get a scalar -- the location of 2223 in column A, which is 2 in 0-based indexing. std::cout third_item.scalar_asarrow::Int64Scalar().value std::endl; // (Doc section: Index Inspection) // (Doc section: Ret) return arrow::Status::OK(); } // (Doc section: Ret) // (Doc section: Main) int main() { arrow::Status st RunMain(); if (!st.ok()) { std::cerr st std::endl; return 1; } return 0; } // (Doc section: Main)代码中的(Doc section: ...)注释标记是文档生成用的分节锚点不影响编译可原样保留也可删除。三步拆解 compute 调用第 1 步构建输入数据并初始化 compute 模块在 compute_example.cc 的 Create Tables 段里程序先用Int32Builder构造两个 int32 数组some_nums为 A 列more_nums为 B 列再用arrow::Table::Make(schema, {some_nums, more_nums}, 5)组装成一张两列表行数为 5。紧接着这一行是本文场景的关键ARROW_RETURN_NOT_OK(arrow::compute::Initialize());compute.rst 明确说明compute 库要求调用arrow::compute::Initialize()才会把各个函数注册进全局FunctionRegistry不调用它的话只有 Arrow 核心功能所需的函数可用而这套核心函数集合是库的实现细节、不应视为稳定接口。所以只要你的程序要用 compute 函数名做查找CallFunction这一步就是必做的。第 2 步用便捷函数compute::Sum对一列求和求和这类高频操作有专门的便捷函数arrow::compute::Sum它把CallFunction的复杂度藏了起来。输入是一个由Datum组成的容器{table-GetColumnByName(A)}利用std::shared_ptrChunkedArray到Datum的隐式转换结果写入一个预声明的arrow::Datum sumARROW_ASSIGN_OR_RAISE(sum, arrow::compute::Sum({table-GetColumnByName(A)}));arrow::Datum是 compute 函数统一的输入输出容器可以是Scalar、Array、ChunkedArray等不同形状所以拿到结果后不能直接打印要先看它装着什么再按对应类型取出// 查看 Datum 类型和内容类型 std::cout Datum kind: sum.ToString() content type: sum.type()-ToString() std::endl; // 显式请求标量取回值 std::cout sum.scalar_asarrow::Int64Scalar().value std::endl;求和结果的Datum里是一个 int64 的Scalar教程给出的输出值为 12891文档示例输出。如果某个函数是否存在便捷入口拿不准去 Compute API 参考api/compute查一下该函数名即可。第 3 步用CallFunction做两列逐元素相加Sum掩盖的那一层就是compute::CallFunction第一个参数是函数名字符串第二个参数是Datum输入组成的 vector可选的第三个参数是 Options 结构体指针。对 A、B 两列做逐元素加法ARROW_ASSIGN_OR_RAISE(element_wise_sum, arrow::compute::CallFunction( add, {table-GetColumnByName(A), table-GetColumnByName(B)}));这个结果Datum装的是 int32 的ChunkedArray用chunked_array()-ToString()打印。教程给出的输出文档示例输出Datum kind: ChunkedArray content type: int32 [ [ 75376, 647, 2287, 5671, 5092 ] ]函数名与输入/输出类型的完整对照见 compute.rst 的 Available functions 一节文中锚点compute-function-list它是配合CallFunction查名字的地方。第 4 步带 Options 结构的index查找有些函数除了数据输入还需要额外参数这时用一个 Options 结构体承载。以在 A 列中查找值 2223该列第 3 个元素为例index函数需要一个compute::IndexOptionsarrow::compute::IndexOptions index_options; // We need an Arrow Scalar, not a raw value. index_options.value arrow::MakeScalar(2223); ARROW_ASSIGN_OR_RAISE( third_item, arrow::compute::CallFunction(index, {table-GetColumnByName(A)}, index_options));注意IndexOptions::value是一个 ArrowScalar而不是裸值所以用arrow::MakeScalar(2223)构造。结果同样先检查Datum类型这里是 int64 的Scalar再取.value打印输出为 2即 2223 在 A 列中的 0 起始下标文档示例输出。结果验证与排查运行后怎么算成功main()的模式是RunMain()返回arrow::Status非 OK 时打印状态并返回 1因此退出码为 0 且三处std::cout输出与上文示例一致即完成。程序末尾显式return arrow::Status::OK();。函数调用返回错误时怎么看所有 compute 调用都走ARROW_ASSIGN_OR_RAISEStatus不是 OK 就会在main()里被打印出来。compute.rst 给出的可预期行为有两条排查时可对照不支持的输入类型返回TypeError的Status。文档原话建议如果不确定某个函数是否支持具体输入类型就试一下。二元数值运算在两个输入类型不同时会先按 Common numeric type 规则提升到能容纳两边所有值的公共类型再执行文档特别提醒uint64列与int16列比较时若某个uint64值无法表示为公共类型int64例如2 ** 63会发出错误。输入形状注意各函数接受的输入是Datum形状的 tagged union。多数函数同时支持标量与chunked数组但也有例外——比如array_sort_indices要求输入必须是 array而通用的sort_indices可以接受 array、chunked array、record batch 或 table。用CallFunction报类型相关错误时先对照函数列表里该函数的 Input types 列确认你传的形状是否被接受。使用边界Grouped Aggregations 不能走CallFunctioncompute.rst 明确写了 Grouped Aggregations are not invocable viaCallFunction。hash_sum这类hash_前缀函数是 SQL 风格 group by 的内部组成部分本文的通用调用路径覆盖不到它们。算术函数默认不检测溢出add等默认变体溢出时回绕需要用add_checked这类_checked变体溢出时返回Invalid的Status。隐式 castkernel 与参数类型不完全匹配时函数可以做隐式转换例如比较字典编码数组时会先解码再比较各函数自行定义其转换行为写比较和算术逻辑时以 Implicit casts 一节的规则为准。想继续深入具体函数签名时入口是 Compute API 参考api/compute教程结尾也指出当数据超出内存时下一步是通过 Arrow Datasets 处理 larger-than-memory 数据集但那已经超出本文的 compute 调用范围。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考