读取和写入 CSV 文件#
Arrow 提供了一个快速的 CSV 读取器,允许将外部数据导入以创建 Arrow 表格或 Arrow RecordBatch 流。
另请参阅
读取 CSV 文件#
CSV 文件中的数据可以作为单个 Arrow 表格使用 TableReader
读取,也可以使用 StreamingReader
以 RecordBatch 流的形式读取。有关这两种方法之间权衡的讨论,请参阅 权衡。
这两个读取器都需要一个表示输入文件的 arrow::io::InputStream
实例。可以使用 ReadOptions
、ParseOptions
和 ConvertOptions
的组合来自定义其行为。
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
}
}
权衡#
使用 TableReader
或 StreamingReader
的选择最终将取决于用例,但需要注意一些权衡
内存使用情况:
TableReader
会一次性将所有数据加载到内存中,并且根据数据量,可能比StreamingReader
需要更多的内存,后者每次只加载一个RecordBatch
。这可能是用户面临的最重要的权衡。速度:读取 CSV 的全部内容时,
TableReader
往往比StreamingReader
快,因为它更好地利用了可用的内核。有关更多详细信息,请参阅 性能。灵活性:
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_names
和 ConvertOptions::include_columns
,则假定 ReadOptions::column_names
映射到 CSV 列,而 ConvertOptions::include_columns
是这些列名中将成为 Arrow 表格一部分的子集。
数据类型#
默认情况下,CSV 读取器会为每列推断最合适的数据类型。类型推断会按以下顺序考虑以下数据类型:
空值
Int64
布尔值
Date32
Time32(以秒为单位)
Timestamp(以秒为单位)
Timestamp(以纳秒为单位)
Float64
Dictionary<String>(如果
ConvertOptions::auto_dict_encode
为真)Dictionary<Binary>(如果
ConvertOptions::auto_dict_encode
为真)字符串
二进制
可以通过设置 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 |
结果类型 |
---|---|---|
(推断) |
|
timestamp[s] |
|
timestamp[s, UTC] |
|
|
||
2021-01-01T00:00:00
2021-01-01T00:00:00Z
|
字符串 |
|
timestamp[s] |
|
timestamp[s] |
|
(错误) |
|
|
||
2021-01-01T00:00:00
2021-01-01T00:00:00Z
|
||
timestamp[s, UTC] |
|
(错误) |
|
timestamp[s, UTC] |
|
|
||
2021-01-01T00:00:00
2021-01-01T00:00:00Z
|
(错误) |
|
timestamp[s, America/New_York] |
|
(错误) |
|
timestamp[s, America/New_York] |
|
|
||
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 数据字节)。