Arrow Datasets 

Arrow C++提供了数据集（Datasets）的概念和实现，用于处理分段的数据，这些数据可以超出内存大小，可能是由于生成大量数据、从流中读取数据或在磁盘上具有大型文件。在本文中，您将学到如何：

• 读取一个多文件分区数据集并将其放入一个Table中。

• 从Table中写出一个分区数据集。

先决条件 

在继续之前，请确保您具备以下条件：

• 已安装Arrow，您可以在此处设置：在您自己的项目中使用Arrow C++。

• 对Arrow的基本数据结构有一定的了解，可参考基本的Arrow数据结构。

为了更好地理解，可能还有必要阅读Arrow文件I/O。但这不是必需的。

设置 

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

• 我们需要包括必要的头文件。

• 需要一个main()函数来将所有部分连接在一起。

• 我们需要一些磁盘上的数据来进行操作。

包括 

在编写C++代码之前，我们需要一些包括。我们将引入用于输出的iostream，然后导入parquet和arrow，以便在本文中使用的每种文件类型：

#include <arrow/api.h>
#include <arrow/dataset/api.h>
// 我们使用Parquet头文件来设置示例；它们不是使用数据集所必需的。
#include <parquet/arrow/reader.h>
#include <parquet/arrow/writer.h>

#include <iostream>

主函数 

为了将所有部分连接在一起，我们将使用先前介绍的基本数据结构教程中的main()模式：

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

与之前的用法类似，此主函数与RunMain()函数配对使用：

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

生成用于读取的文件 

我们需要一些实际操作的文件。在实践中，您可能会有一些用于自己的应用程序的输入数据。然而，在这里，我们希望探索而不需要提供或查找数据集，因此让我们生成一些数据以便于后续的演示。可以阅读下面的内容，但本文将更详细地讨论这些概念。现在，只需将其复制并意识到最后生成了一个分区数据集在磁盘上：

// 为本示例生成一些数据。
arrow::Result<std::shared_ptr<arrow::Table>> CreateTable() {
  // 此代码应该看起来与基本的Arrow示例相似，不是本示例的重点。然而，我们需要数据来处理，这就是数据生成的目的！
  auto schema =
      arrow::schema({arrow::field("a", arrow::int64()), arrow::field("b", arrow::int64()),
                     arrow::field("c", arrow::int64())});
  std::shared_ptr<arrow::Array> array_a;
  std::shared_ptr<arrow::Array> array_b;
  std::shared_ptr<arrow::Array> array_c;
  arrow::NumericBuilder<arrow::Int64Type> builder;
  ARROW_RETURN_NOT_OK(builder.AppendValues({0, 1, 2, 3, 4, 5, 6, 7, 8, 9}));
  ARROW_RETURN_NOT_OK(builder.Finish(&array_a));
  builder.Reset();
  ARROW_RETURN_NOT_OK(builder.AppendValues({9, 8, 7, 6, 5, 4, 3, 2, 1, 0}));
  ARROW_RETURN_NOT_OK(builder.Finish(&array_b));
  builder.Reset();
  ARROW_RETURN_NOT_OK(builder.AppendValues({1, 2, 1, 2, 1, 2, 1, 2, 1, 2}));
  ARROW_RETURN_NOT_OK(builder.Finish(&array_c));
  return arrow::Table::Make(schema, {array_a, array_b, array_c});
}

// 通过写入两个Parquet文件来设置一个数据集。
arrow::Result<std::string> CreateExampleParquetDataset(
    const std::shared_ptr<arrow::fs::FileSystem>& filesystem,
    const std::string& root_path) {
  // 与CreateTable()类似，这是帮助我们读取的数据集。不用担心，我们在示例中也会写入一个数据集。
  auto base_path = root_path + "parquet_dataset";
  ARROW_RETURN_NOT_OK(filesystem->CreateDir(base_path));
  // 创建一个Arrow Table
  ARROW_ASSIGN_OR_RAISE(auto table, CreateTable());
  // 将其写入两个Parquet文件
  ARROW_ASSIGN_OR_RAISE(auto output,
                        filesystem->OpenOutputStream(base_path + "/data1.parquet"));
  ARROW_RETURN_NOT_OK(parquet::arrow::WriteTable(
      *table->Slice(0, 5), arrow::default_memory_pool(), output, 2048));
  ARROW_ASSIGN_OR_RAISE(output,
                        filesystem->OpenOutputStream(base_path + "/data2.parquet"));
  ARROW_RETURN_NOT_OK(parquet::arrow::WriteTable(
      *table->Slice(5), arrow::default_memory_pool(), output, 2048));
  return base_path;
}

arrow::Status PrepareEnv() {
  // 通过设置一些快速的写入来为我们的环境准备好以供读取。
  ARROW_ASSIGN_OR_RAISE(auto src_table, CreateTable())
  std::shared_ptr<arrow::fs::FileSystem> setup_fs;
  // 请注意，此操作在可执行文件构建的目录中执行。
  char setup_path[256];
  char* result = getcwd(setup_path, 256);
  if (result == NULL) {
    return arrow::Status::IOError("获取PWD失败。");
  }

  ARROW_ASSIGN_OR_RAISE(setup_fs, arrow::fs::FileSystemFromUriOrPath(setup_path));
  ARROW_ASSIGN_OR_RAISE(auto dset_path, CreateExampleParquetDataset(setup_fs, ""));

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

为了确实生成这些文件，确保RunMain()函数中的第一个调用是我们的辅助函数PrepareEnv()，它将为我们提供磁盘上的一个数据集：

  ARROW_RETURN_NOT_OK(PrepareEnv());

读取分区数据集 

读取数据集与读取单个文件是不同的任务。这个任务需要比读取单个文件更多的工作，因为它需要能够解析多个文件和/或文件夹。这个过程可以分为以下步骤：

• 获取用于本地文件系统的fs::FileSystem对象

• 创建一个fs::FileSelector并使用它来准备一个dataset::FileSystemDatasetFactory

• 使用dataset::FileSystemDatasetFactory构建一个dataset::Dataset

• 使用dataset::Scanner读入一个Table

为了开始，我们需要能够与本地文件系统进行交互。为了做到这一点，我们需要一个 fs::FileSystem 对象。fs::FileSystem 是一个抽象，允许我们使用相同的接口，无论是使用Amazon S3、Google Cloud Storage还是本地磁盘，而我们将使用本地磁盘。所以，让我们声明它：

// 首先，我们需要一个文件系统对象，它允许我们与本地文件系统交互，从给定的路径开始。为了简单起见，这将是当前目录。
std::shared_ptr<arrow::fs::FileSystem> fs;

在本例中，我们的文件系统的基本路径将存在于与可执行文件相同的目录中。fs::FileSystemFromUriOrPath() 允许我们获取适用于支持的文件系统类型的 fs::FileSystem 对象。在这里，我们只需传递我们的路径：

// 获取当前工作目录并使用它来创建FileSystem对象。
char init_path[256];
char* result = getcwd(init_path, 256);
if (result == NULL) {
  return arrow::Status::IOError("获取PWD失败。");
}
ARROW_ASSIGN_OR_RAISE(fs, arrow::fs::FileSystemFromUriOrPath(init_path));

创建一个 FileSystemDatasetFactory:

// 获取一个数据集::Dataset的时候需要准备一个数据集::FileSystemDatasetFactory。
//这是一个很长但非常描述性的名称 - 它将为我们提供一个从我们的fs::FileSystem获取数据的工厂。
//首先，我们通过填充一个数据集::FileSystemFactoryOptions结构来配置它：

// 创建一个选项对象允许我们配置数据集读取。
arrow::dataset::FileSystemFactoryOptions options;
// 我们将使用Hive风格的分区。我们将让Arrow Datasets推断分区模式。我们不会设置其他选项，使用默认值就足够了。
options.partitioning = arrow::dataset::HivePartitioning::MakeFactory();

有很多文件格式可用，我们必须选择一个在实际读取时会被期望的文件格式。由于磁盘上有Parquet，所以当我们进行读取时，当然会要求Parquet：

auto read_format = std::make_shared<arrow::dataset::ParquetFileFormat>();

设置好 fs::FileSystem、fs::FileSelector、选项和文件格式后，我们可以创建 FileSystemDatasetFactory。这只需要将我们准备的所有内容传递并分配给一个变量：

// 现在，我们得到一个工厂，允许我们获取我们的数据集 - 我们还没有数据集！
ARROW_ASSIGN_OR_RAISE(auto factory, arrow::dataset::FileSystemDatasetFactory::Make(
                                      fs, selector, read_format, options));

使用工厂构建数据集：

// 现在，我们从工厂构建数据集。
ARROW_ASSIGN_OR_RAISE(auto read_dataset, factory->Finish());

现在，我们在内存中有一个 Dataset 对象。这并不意味着整个数据集都在内存中，但我们现在可以访问工具，允许我们探索和使用磁盘上的数据集。例如，我们可以获取构成整个数据集的片段（文件），并打印出来，以及一些小信息：

// 打印出片段
ARROW_ASSIGN_OR_RAISE(auto fragments, read_dataset->GetFragments());
for (const auto& fragment : fragments) {
  std::cout << "找到片段：" << (*fragment)->ToString() << std::endl;
  std.	cout << "分区表达式：" << (*fragment)->partition_expression().ToString() << std::endl;
}

将数据集移入表格：

// 一个处理数据集的方法是将它们转换为表格，其中我们可以对表格进行的操作与之前我们了解的操作相同。
// 将数据集扫描为表格，完成后，您可以对其执行常规表格操作，如计算和打印。不过，现在您也需要在内存中运行。
ARROW_ASSIGN_OR_RAISE(auto read_scan_builder, read_dataset->NewScan());
  // (文档部分: 读取扫描生成器)
  // (文档部分: 读取扫描器)
ARROW_ASSIGN_OR_RAISE(auto read_scanner, read_scan_builder->Finish());
// 获取数据集进入表格
ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Table> table, read_scanner->ToTable());
std::cout << table->ToString();

从表格将数据集写入磁盘：

// 写入数据集::Dataset与写入单个文件不同。这项任务比写入单个文件需要更多的工作，因为需要能够处理跨多个文件和文件夹的分区方案。这个过程可以分为以下步骤：

// 准备TableBatchReader

// 创建一个数据集::Scanner来从TableBatchReader中提取数据

// 准备模式、分区和文件格式选项

// 配置数据集::FileSystemDatasetWriteOptions - 一个配置写入函数的结构

// 将数据集写入磁盘

// 准备从表格写入的数据

// 我们有一个表格，想要将数据集::Dataset写入磁盘。事实上，为了进行探索，我们将使用不同的分区方案，而不仅仅是像原始片段那样划分为两半，
//而是根据每行在 "a" 列中的值进行分区。

// 为了开始这项任务，让我们获取一个TableBatchReader！这样很容易将数据写入数据集，并且可以在其他地方使用，每当需要将表格分解成RecordBatches的流时，
//我们可以使用它。在这里，我们只需使用TableBatchReader的构造函数，传入我们的表格：

// 获取一个Table以进行写入准备
std::shared_ptr<arrow::TableBatchReader> write_dataset =
std::make_shared<arrow::TableBatchReader>(table);

创建用于移动表格数据的 Scanner：

// 写入数据集::Dataset的过程，一旦有数据源可用，与读取相反。在之前，我们使用了一个数据集::Scanner，以便将数据扫描到表格中 - 现在，我们需要一个数据集::Scanner，以从我们的TableBatchReader中读取数据。为了获取数据集::Scanner，我们将基于我们的TableBatchReader创建一个数据集::ScannerBuilder，然后使用该构建器构建数据集::Scanner：

auto write_scanner_builder =
  arrow::dataset::ScannerBuilder::FromRecordBatchReader(write_dataset);
ARROW_ASSIGN_OR_RAISE(auto write_scanner, write_scanner_builder->Finish())

准备模式、分区和文件格式选项：

// 由于我们想根据 "a" 列分区，我们需要声明这一点。在定义我们的分区模式时，我们只有一个包含 "a" 的字段：

// 分区模式确定用于分区的键是哪些字段。
auto partition_schema = arrow::schema({arrow::field("a", arrow::utf8())});

选择分区算法，我们将再次使用 Hive 风格分区，这次将模式传递给它作为配置：

// 我们将使用 Hive 风格分区，这将创建带有 "key=value" 对的目录。
auto partitioning = std::make_shared<arrow::dataset::HivePartitioning>(partition_schema);

选择文件格式，通常使用 Parquet：

// 现在，我们声明我们将写入Parquet文件。
auto write_format = std::make_shared<arrow::dataset::ParquetFileFormat>();

配置 FileSystemDatasetWriteOptions：

// 为了将数据写入磁盘，我们需要一些配置。我们将通过在一个`dataset::FileSystemDatasetWriteOptions`结构中设置值来完成这项任务。我们将尽量使用默认值进行初始化：

// 这一次，我们创建用于写入的选项，但会进行更多配置。
arrow::dataset::FileSystemDatasetWriteOptions write_options;
// 默认值开始。
write_options.file_write_options = write_format->DefaultWriteOptions();

目标文件系统：

// 目标是使用我们已经有的文件系统。
write_options.filesystem = fs;

目标目录：

// 写入到当前目录中的 "write_dataset" 文件夹。
write_options.base_dir = "write_dataset";

分区方法：

// 使用上面声明的分区。
write_options.partitioning = partitioning;

文件名称模板：

// 定义构成数据集的文件的名称。
write_options.basename_template = "part{i}.parquet";

处理已存在的数据的行为：

// 设置行为以覆盖现有数据 - 具体来说，这使得可以多次运行此示例，并允许您的代码覆盖已有的内容。
write_options.existing_data_behavior =
  arrow::dataset::ExistingDataBehavior::kOverwriteOrIgnore;

将数据集写入磁盘：

// 一旦配置了`dataset::FileSystemDatasetWriteOptions`，并且准备好了`dataset::Scanner`来解析数据，我们可以将选项和`dataset::Scanner`传递给`dataset::FileSystemDataset::Write()`，将数据写入磁盘：

ARROW_RETURN_NOT_OK(
  arrow::dataset::FileSystemDataset::Write(write_options, write_scanner));

最后，返回 Status::OK()，以便 main() 知道我们已经完成了，一切正常，就像前面的教程一样：

return arrow::Status::OK();

这样，您已经读取和写入了分区数据集！这种方法在进行任何支持的数据集格式时，经过一些配置，都可以工作。对于这样一个数据集的示例，纽约出租车数据集是一个众所周知的数据集，您可以在这里找到。

这意味着现在我们必须能够处理这些数据，而不需要一次将所有数据都加载到内存中。对于这一点，可以尝试使用 Acero。