读取和写入 CSV 文件#

Arrow 提供了一个快速的 CSV 读取器,允许将外部数据导入以创建 Arrow 表格或 Arrow RecordBatch 流。

读取 CSV 文件#

CSV 文件中的数据可以作为单个 Arrow 表格使用 TableReader 读取,也可以使用 StreamingReader 以 RecordBatch 流的形式读取。有关这两种方法之间权衡的讨论,请参阅 权衡

这两个读取器都需要一个表示输入文件的 arrow::io::InputStream 实例。可以使用 ReadOptionsParseOptionsConvertOptions 的组合来自定义其行为。

TableReader#

#include "arrow/csv/api.h"

{
   // ...
   arrow::io::IOContext io_context = arrow::io::default_io_context();
   std::shared_ptr<arrow::io::InputStream> input = ...;

   auto read_options = arrow::csv::ReadOptions::Defaults();
   auto parse_options = arrow::csv::ParseOptions::Defaults();
   auto convert_options = arrow::csv::ConvertOptions::Defaults();

   // Instantiate TableReader from input stream and options
   auto maybe_reader =
     arrow::csv::TableReader::Make(io_context,
                                   input,
                                   read_options,
                                   parse_options,
                                   convert_options);
   if (!maybe_reader.ok()) {
     // Handle TableReader instantiation error...
   }
   std::shared_ptr<arrow::csv::TableReader> reader = *maybe_reader;

   // Read table from CSV file
   auto maybe_table = reader->Read();
   if (!maybe_table.ok()) {
     // Handle CSV read error
     // (for example a CSV syntax error or failed type conversion)
   }
   std::shared_ptr<arrow::Table> table = *maybe_table;
}

StreamingReader#

#include "arrow/csv/api.h"

{
   // ...
   arrow::io::IOContext io_context = arrow::io::default_io_context();
   std::shared_ptr<arrow::io::InputStream> input = ...;

   auto read_options = arrow::csv::ReadOptions::Defaults();
   auto parse_options = arrow::csv::ParseOptions::Defaults();
   auto convert_options = arrow::csv::ConvertOptions::Defaults();

   // Instantiate StreamingReader from input stream and options
   auto maybe_reader =
     arrow::csv::StreamingReader::Make(io_context,
                                       input,
                                       read_options,
                                       parse_options,
                                       convert_options);
   if (!maybe_reader.ok()) {
     // Handle StreamingReader instantiation error...
   }
   std::shared_ptr<arrow::csv::StreamingReader> reader = *maybe_reader;

   // Set aside a RecordBatch pointer for re-use while streaming
   std::shared_ptr<RecordBatch> batch;

   while (true) {
       // Attempt to read the first RecordBatch
       arrow::Status status = reader->ReadNext(&batch);

       if (!status.ok()) {
         // Handle read error
       }

       if (batch == NULL) {
         // Handle end of file
         break;
       }

       // Do something with the batch
   }
}

权衡#

使用 TableReaderStreamingReader 的选择最终将取决于用例,但需要注意一些权衡

  1. 内存使用情况:TableReader 会一次性将所有数据加载到内存中,并且根据数据量,可能比 StreamingReader 需要更多的内存,后者每次只加载一个 RecordBatch。这可能是用户面临的最重要的权衡。

  2. 速度:读取 CSV 的全部内容时,TableReader 往往比 StreamingReader 快,因为它更好地利用了可用的内核。有关更多详细信息,请参阅 性能

  3. 灵活性:StreamingReader 可能被认为不如 TableReader 灵活性,因为它仅对读取的第一个块执行类型推断,此后类型将被冻结,并且任何后续块中无法转换为这些类型的数据都将导致错误。请注意,可以通过将 ReadOptions::block_size 设置为足够大的值或使用 ConvertOptions::column_types 显式设置所需的数据类型来解决此问题。

写入 CSV 文件#

CSV 文件写入 OutputStream

#include <arrow/csv/api.h>
{
    // Oneshot write
    // ...
    std::shared_ptr<arrow::io::OutputStream> output = ...;
    auto write_options = arrow::csv::WriteOptions::Defaults();
    if (WriteCSV(table, write_options, output.get()).ok()) {
        // Handle writer error...
    }
}
{
    // Write incrementally
    // ...
    std::shared_ptr<arrow::io::OutputStream> output = ...;
    auto write_options = arrow::csv::WriteOptions::Defaults();
    auto maybe_writer = arrow::csv::MakeCSVWriter(output, schema, write_options);
    if (!maybe_writer.ok()) {
        // Handle writer instantiation error...
    }
    std::shared_ptr<arrow::ipc::RecordBatchWriter> writer = *maybe_writer;

    // Write batches...
    if (!writer->WriteRecordBatch(*batch).ok()) {
        // Handle write error...
    }

    if (!writer->Close().ok()) {
        // Handle close error...
    }
    if (!output->Close().ok()) {
        // Handle file close error...
    }
}

注意

写入器尚不支持所有 Arrow 类型。

列名#

有三种可能的方法可以从 CSV 文件中推断列名

  • 默认情况下,列名从 CSV 文件的第一行读取

  • 如果设置了 ReadOptions::column_names,它会强制表格中的列名采用这些值(CSV 文件的第一行作为数据读取)

  • 如果 ReadOptions::autogenerate_column_names 为真,则列名将使用模式“f0”、“f1”……自动生成(CSV 文件的第一行作为数据读取)

列选择#

默认情况下,Arrow 读取 CSV 文件中的所有列。您可以使用 ConvertOptions::include_columns 选项缩小列的选择范围。如果 ConvertOptions::include_columns 中的一些列在 CSV 文件中不存在,则会发出错误,除非 ConvertOptions::include_missing_columns 为真,在这种情况下,假定缺失的列包含全空值。

与列名的交互#

如果同时指定了 ReadOptions::column_namesConvertOptions::include_columns,则假定 ReadOptions::column_names 映射到 CSV 列,而 ConvertOptions::include_columns 是这些列名中将成为 Arrow 表格一部分的子集。

数据类型#

默认情况下,CSV 读取器会为每列推断最合适的数据类型。类型推断会按以下顺序考虑以下数据类型:

可以通过设置 ConvertOptions::column_types 选项来覆盖选定列的类型推断。可以从以下列表中选择显式数据类型:

  • 空值

  • 所有整数类型

  • Float32 和 Float64

  • Decimal128

  • 布尔值

  • Date32 和 Date64

  • Time32 和 Time64

  • Timestamp

  • Binary 和 Large Binary

  • String 和 Large String(带可选的 UTF8 输入验证)

  • 固定大小的二进制

  • 字典,索引类型为 Int32,值类型为以下之一:Binary、String、LargeBinary、LargeString、Int32、UInt32、Int64、UInt64、Float32、Float64、Decimal128

其他数据类型不支持从 CSV 值转换,并将出错。

字典推断#

如果启用了类型推断并且 ConvertOptions::auto_dict_encode 为真,则 CSV 读取器首先尝试将类似字符串的列转换为字典编码的类似字符串的数组。当达到 ConvertOptions::auto_dict_max_cardinality 中的阈值时,它会切换到普通的类似字符串的数组。

Timestamp 推断/解析#

如果启用了类型推断,则 CSV 读取器首先尝试将类似字符串的列解释为时间戳。如果所有行都有一些时区偏移量(例如 Z+0100),即使偏移量不一致,则推断的类型也将是 UTC 时间戳。如果没有任何行具有时区偏移量,则推断的类型将是无时区的时间戳。混合使用有/无偏移量的行将导致字符串列。

如果类型被显式指定为有/无时区的时间戳,则读取器将在该列中遇到没有/有时区偏移量的值时出错。请注意,这意味着目前无法让读取器将时间戳列(无时区)解析为特定时区的本地时间;而是将该列解析为无时区的时间戳,然后使用 assume_timezone 计算函数在之后转换值。

指定类型

输入 CSV

结果类型

(推断)

2021-01-01T00:00:00

timestamp[s]

2021-01-01T00:00:00Z

timestamp[s, UTC]

2021-01-01T00:00:00+0100

2021-01-01T00:00:00
2021-01-01T00:00:00Z

字符串

timestamp[s]

2021-01-01T00:00:00

timestamp[s]

2021-01-01T00:00:00Z

(错误)

2021-01-01T00:00:00+0100

2021-01-01T00:00:00
2021-01-01T00:00:00Z

timestamp[s, UTC]

2021-01-01T00:00:00

(错误)

2021-01-01T00:00:00Z

timestamp[s, UTC]

2021-01-01T00:00:00+0100

2021-01-01T00:00:00
2021-01-01T00:00:00Z

(错误)

timestamp[s, America/New_York]

2021-01-01T00:00:00

(错误)

2021-01-01T00:00:00Z

timestamp[s, America/New_York]

2021-01-01T00:00:00+0100

2021-01-01T00:00:00
2021-01-01T00:00:00Z

(错误)

空值#

空值根据存储在 ConvertOptions::null_values 中的拼写识别。 ConvertOptions::Defaults() 工厂方法将初始化许多常规的空值拼写,例如 N/A

字符编码#

CSV 文件应使用 UTF8 编码。但是,对于 Binary 列,接受非 UTF8 数据。

写入选项#

可以通过 WriteOptions 自定义写入的 CSV 文件的格式。目前可用选项很少;将在将来的版本中添加更多选项。

性能#

默认情况下,TableReader 将并行化读取以利用机器上的所有 CPU 内核。您可以在 ReadOptions::use_threads 中更改此设置。在性能优良的台式机或笔记本电脑上,合理的预期是每个内核至少 100 MB/s(以源 CSV 字节为单位测量,而不是目标 Arrow 数据字节)。