Arrow 计算#

Apache Arrow 提供计算函数以促进高效和便携的数据处理。本文中，您将使用 Arrow 的计算功能来

计算列的总和
计算两列的元素级总和
在列中搜索值

先决条件#

在继续之前，请确保您已具备：

Arrow 的安装，您可以在此处设置：在您自己的项目中使用 Arrow C++。如果您是自行编译 Arrow，请确保您在编译时启用了计算模块（即 -DARROW_COMPUTE=ON），请参阅可选组件。
通过基本的 Arrow 数据结构了解 Arrow 基本数据结构

设置#

在运行一些计算之前，我们需要填补几个空白

我们需要包含必要的头文件。
需要一个 main() 函数来将所有内容连接起来。
我们需要数据来操作。

包含文件#

在编写 C++ 代码之前，我们需要一些头文件。我们将获取 iostream 用于输出，然后导入 Arrow 的计算功能

#include <arrow/api.h>
#include <arrow/compute/api.h>

#include <iostream>

Main()#

对于连接代码，我们将使用之前关于数据结构教程中的 main() 模式。

int main() {
  arrow::Status st = RunMain();
  if (!st.ok()) {
    std::cerr << st << std::endl;
    return 1;
  }
  return 0;
}

它与 RunMain() 搭配使用，就像我们之前使用它一样。

arrow::Status RunMain() {

为计算生成表格#

在我们开始之前，我们将初始化一个带有两列的 Table 来操作。我们将使用基本 Arrow 数据结构中的方法，所以如果有什么令人困惑的地方，请回头查看

  // 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_ptr<arrow::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_ptr<arrow::Array> more_nums;
  ARROW_ASSIGN_OR_RAISE(more_nums, int32builder.Finish());

  // Make a table out of our pair of arrays.
  std::shared_ptr<arrow::Field> field_a, field_b;
  std::shared_ptr<arrow::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_ptr<arrow::Table> table;
  table = arrow::Table::Make(schema, {some_nums, more_nums}, 5);

计算数组的总和#

使用计算函数有两个一般步骤，我们在此处将其分开

准备一个用于输出的 Datum
调用 compute::Sum()，一个用于计算 Array 总和的便利函数
检索并打印输出

使用 Datum 准备输出内存#

计算完成后，我们需要一个地方来存放结果。在 Arrow 中，用于此类输出的对象称为 Datum。此对象用于在计算函数中传递输入和输出，并且可以包含许多不同形状的 Arrow 数据结构。我们需要它来检索计算函数的输出。

  // The Datum class is what all compute functions output to, and they can take Datums
  // as inputs, as well.
  arrow::Datum sum;

调用 Sum()#

在这里，我们将获取我们的 Table，它有“A”和“B”两列，并对列“A”求和。对于求和，有一个便利函数 compute::Sum()，它减少了计算接口的复杂性。我们将在下一次计算中查看更复杂的版本。对于给定函数，请参阅计算函数以查看是否存在便利函数。compute::Sum() 接受一个给定的 Array 或 ChunkedArray – 在这里，我们使用 Table::GetColumnByName() 传入列 A。然后，它将输出到 Datum。将所有这些放在一起，我们得到

  // Here, we can use arrow::compute::Sum. This is a convenience function, and the next
  // computation won't be so simple. However, using these where possible helps
  // readability.
  ARROW_ASSIGN_OR_RAISE(sum, arrow::compute::Sum({table->GetColumnByName("A")}));

从 Datum 获取结果#

上一步我们得到了一个包含我们总和的 Datum。但是，我们不能直接打印它——它在保存任意 Arrow 数据结构方面的灵活性意味着我们必须小心地检索数据。首先，要了解它里面有什么，我们可以检查它是什么类型的数据结构，然后它包含什么类型的原始数据

  // 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;

这应该报告 Datum 存储一个带有 64 位整数的 Scalar。为了查看该值是什么，我们可以像这样打印它，它会得到 12891

  // 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_as<arrow::Int64Scalar>().value << std::endl;

现在我们已经使用了 compute::Sum() 并从中得到了我们想要的东西！

使用 CallFunction() 计算元素级数组加法#

下一层复杂性使用了 compute::Sum() 巧妙地隐藏的东西：compute::CallFunction()。对于此示例，我们将探讨如何将更健壮的 compute::CallFunction() 与“add”计算函数一起使用。模式仍然相似

准备一个用于输出的 Datum
使用“add”调用 compute::CallFunction()
检索并打印输出

使用 Datum 准备输出内存#

再次，我们需要一个 Datum 来存放我们得到的任何输出

  arrow::Datum element_wise_sum;

使用 CallFunction() 和“add”#

compute::CallFunction() 将所需函数的名称作为其第一个参数，然后将该函数的数据输入作为向量作为其第二个参数。现在，我们希望对“A”列和“B”列进行元素级加法。因此，我们将请求“add”，传入“A 和 B”列，并将输出写入我们的 Datum。将所有这些放在一起，我们得到

  // 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")}));

另请参阅

可用函数，用于与 compute::CallFunction() 一起使用的其他函数列表

从 Datum 获取结果#

再次，Datum 需要一些小心的处理。当我们知道它里面有什么时，处理起来会容易得多。这个 Datum 包含一个带有 32 位整数的 ChunkedArray，但我们可以打印出来确认

  // 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;

由于它是一个 ChunkedArray，我们从 Datum 请求它——ChunkedArray 有一个 ChunkedArray::ToString() 方法，所以我们将用它来打印出其内容

  // This time, we get a ChunkedArray, not a scalar.
  std::cout << element_wise_sum.chunked_array()->ToString() << std::endl;

输出结果如下

Datum kind: ChunkedArray content type: int32
[
  [
    75376,
    647,
    2287,
    5671,
    5092
  ]
]

现在，我们已经使用了 compute::CallFunction()，而不是便利函数！这使得可用的计算范围更广。

使用 CallFunction() 和 Options 搜索值#

一类计算仍然存在。compute::CallFunction() 使用向量作为数据输入，但计算通常需要额外的参数才能运行。为了提供这些，计算函数可以与结构体关联，在其中可以定义它们的参数。您可以在这里查看给定函数使用了哪个结构体。对于此示例，我们将使用“index”计算函数在列“A”中搜索值。此过程有三个步骤，而不是之前的两个

准备一个用于输出的 Datum
准备 compute::IndexOptions
使用“index”和 compute::IndexOptions 调用 compute::CallFunction()
检索并打印输出

使用 Datum 准备输出内存#

我们需要一个 Datum 来存放我们得到的任何输出

  // Use an options struct to set up searching for 2223 in column A (the third item).
  arrow::Datum third_item;

使用 IndexOptions 配置“index”#

对于本次探索，我们将使用“index”函数——这是一种搜索方法，它返回输入值的索引。为了传递此输入值，我们需要一个 compute::IndexOptions 结构体。所以，让我们创建那个结构体

  // An options struct is used in lieu of passing an arbitrary amount of arguments.
  arrow::compute::IndexOptions index_options;

在搜索函数中，需要一个目标值。在这里，我们将使用列 A 中的第三项 2223，并相应地配置我们的结构体

  // We need an Arrow Scalar, not a raw value.
  index_options.value = arrow::MakeScalar(2223);

使用 CallFunction() 和“index”以及 IndexOptions#

为了实际运行函数，我们再次使用 compute::CallFunction()，这次将我们的 IndexOptions 结构体作为第三个参数按引用传递。和以前一样，第一个参数是函数名，第二个是我们的数据输入

  ARROW_ASSIGN_OR_RAISE(
      third_item, arrow::compute::CallFunction("index", {table->GetColumnByName("A")},
                                               &index_options));

从 Datum 获取结果#

最后一次，让我们看看我们的 Datum 有什么！这将是一个带有 64 位整数的 Scalar，输出将是 2

  // 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_as<arrow::Int64Scalar>().value << std::endl;

结束程序#

最后，我们只需返回 arrow::Status::OK()，这样 main() 就知道我们已经完成，并且一切正常，就像之前的教程一样。

  return arrow::Status::OK();
}

至此，您已经使用了属于三种主要类型的计算函数——有和没有便利函数，然后带有 Options 结构体。现在您可以处理您需要的任何 Table，并解决任何适合内存的数据问题！

这意味着现在我们必须在下一篇文章中通过 Arrow Datasets 了解如何处理大于内存的数据集。

请参考以下内容以获取完整的代码副本。

// (Doc section: Includes)
#include <arrow/api.h>
#include <arrow/compute/api.h>

#include <iostream>
// (Doc section: Includes)

// (Doc section: RunMain)
arrow::Status RunMain() {
  // (Doc section: 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_ptr<arrow::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_ptr<arrow::Array> more_nums;
  ARROW_ASSIGN_OR_RAISE(more_nums, int32builder.Finish());

  // Make a table out of our pair of arrays.
  std::shared_ptr<arrow::Field> field_a, field_b;
  std::shared_ptr<arrow::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_ptr<arrow::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 won't 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_as<arrow::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_as<arrow::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)