读取和写入 Parquet 文件

另请参阅

Parquet 读取器和写入器 API 参考.

Parquet 格式 是一种用于复杂数据的高效列式存储格式。Parquet C++ 实现是 Apache Arrow 项目的一部分，并受益于与 Arrow C++ 类和工具的紧密集成。

读取 Parquet 文件

arrow::FileReader 类将数据读入 Arrow 表和记录批。

StreamReader 类允许使用 C++ 输入流方法逐列逐行读取数据。提供这种方法是为了便于使用和类型安全。当然，当必须在文件增量读取和写入时流式传输数据时，它也很有用。

请注意，由于类型检查和列值一次处理一个的事实，StreamReader 的性能不会那么好。

FileReader

要将 Parquet 数据读入 Arrow 结构，请使用 arrow::FileReader。要构造它，需要一个表示输入文件的 ::arrow::io::RandomAccessFile 实例。要一次读取整个文件，请使用 arrow::FileReader::ReadTable()

// #include "arrow/io/api.h"
// #include "parquet/arrow/reader.h"

arrow::MemoryPool* pool = arrow::default_memory_pool();
std::shared_ptr<arrow::io::RandomAccessFile> input;
ARROW_ASSIGN_OR_RAISE(input, arrow::io::ReadableFile::Open(path_to_file));

// Open Parquet file reader
std::unique_ptr<parquet::arrow::FileReader> arrow_reader;
ARROW_ASSIGN_OR_RAISE(arrow_reader, parquet::arrow::OpenFile(input, pool));

// Read entire file as a single Arrow table
std::shared_ptr<arrow::Table> table;
ARROW_RETURN_NOT_OK(arrow_reader->ReadTable(&table));

更细粒度的选项可通过 arrow::FileReaderBuilder 辅助类获得，该类接受 ReaderProperties 和 ArrowReaderProperties 类。

要读取批处理流，请使用 arrow::FileReader::GetRecordBatchReader() 方法来检索 arrow::RecordBatchReader。它将使用在 ArrowReaderProperties 中设置的批处理大小。

// #include "arrow/io/api.h"
// #include "parquet/arrow/reader.h"

arrow::MemoryPool* pool = arrow::default_memory_pool();

// Configure general Parquet reader settings
auto reader_properties = parquet::ReaderProperties(pool);
reader_properties.set_buffer_size(4096 * 4);
reader_properties.enable_buffered_stream();

// Configure Arrow-specific Parquet reader settings
auto arrow_reader_props = parquet::ArrowReaderProperties();
arrow_reader_props.set_batch_size(128 * 1024);  // default 64 * 1024

parquet::arrow::FileReaderBuilder reader_builder;
ARROW_RETURN_NOT_OK(
    reader_builder.OpenFile(path_to_file, /*memory_map=*/false, reader_properties));
reader_builder.memory_pool(pool);
reader_builder.properties(arrow_reader_props);

std::unique_ptr<parquet::arrow::FileReader> arrow_reader;
ARROW_ASSIGN_OR_RAISE(arrow_reader, reader_builder.Build());

std::shared_ptr<::arrow::RecordBatchReader> rb_reader;
ARROW_RETURN_NOT_OK(arrow_reader->GetRecordBatchReader(&rb_reader));

for (arrow::Result<std::shared_ptr<arrow::RecordBatch>> maybe_batch : *rb_reader) {
  // Operate on each batch...
}

另请参阅

要读取多文件数据集或下推过滤器以修剪行组，请参阅 表格数据集。

性能和内存效率

对于远程文件系统，请使用读取合并（预缓冲）来减少 API 调用的次数

auto arrow_reader_props = parquet::ArrowReaderProperties();
reader_properties.set_prebuffer(true);

默认设置通常会调整到良好的性能，但默认情况下并行列解码是关闭的。在 ArrowReaderProperties 的构造函数中启用它

auto arrow_reader_props = parquet::ArrowReaderProperties(/*use_threads=*/true);

如果内存效率比性能更重要，那么

1. 请勿在 parquet::ArrowReaderProperties 中打开读取合并（预缓冲）。

2. 使用 arrow::FileReader::GetRecordBatchReader() 分批读取数据。

3. 在 parquet::ReaderProperties 中打开 enable_buffered_stream。

此外，如果您知道某些列包含许多重复值，则可以将它们作为 字典编码 列读取。这是通过 ArrowReaderProperties 上的 set_read_dictionary 设置启用的。如果文件是使用 Arrow C++ 编写的并且激活了 store_schema，则将自动读取原始 Arrow 模式并将覆盖此设置。

StreamReader

StreamReader 允许使用标准 C++ 输入运算符读取 Parquet 文件，从而确保类型安全。

请注意，类型必须与模式完全匹配，即如果模式字段是无符号 16 位整数，则必须提供 uint16_t 类型。

异常用于 segnal 错误。ParquetException 在以下情况下抛出

• 尝试通过提供不正确的类型来读取字段。

• 尝试读取超出行尾。

• 尝试读取超出文件尾。

#include "arrow/io/file.h"
#include "parquet/stream_reader.h"

{
   std::shared_ptr<arrow::io::ReadableFile> infile;

   PARQUET_ASSIGN_OR_THROW(
      infile,
      arrow::io::ReadableFile::Open("test.parquet"));

   parquet::StreamReader stream{parquet::ParquetFileReader::Open(infile)};

   std::string article;
   float price;
   uint32_t quantity;

   while ( !stream.eof() )
   {
      stream >> article >> price >> quantity >> parquet::EndRow;
      // ...
   }
}

写入 Parquet 文件

WriteTable

arrow::WriteTable() 函数将整个 ::arrow::Table 写入输出文件。

// #include "parquet/arrow/writer.h"
// #include "arrow/util/type_fwd.h"
using parquet::ArrowWriterProperties;
using parquet::WriterProperties;

ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Table> table, GetTable());

// Choose compression
std::shared_ptr<WriterProperties> props =
    WriterProperties::Builder().compression(arrow::Compression::SNAPPY)->build();

// Opt to store Arrow schema for easier reads back into Arrow
std::shared_ptr<ArrowWriterProperties> arrow_props =
    ArrowWriterProperties::Builder().store_schema()->build();

std::shared_ptr<arrow::io::FileOutputStream> outfile;
ARROW_ASSIGN_OR_RAISE(outfile, arrow::io::FileOutputStream::Open(path_to_file));

ARROW_RETURN_NOT_OK(parquet::arrow::WriteTable(*table.get(),
                                               arrow::default_memory_pool(), outfile,
                                               /*chunk_size=*/3, props, arrow_props));

注意

默认情况下，C++ 中的列压缩是关闭的。请参阅下文，了解如何在写入器属性中选择压缩编解码器。

要逐批写入数据，请使用 arrow::FileWriter。

// #include "parquet/arrow/writer.h"
// #include "arrow/util/type_fwd.h"
using parquet::ArrowWriterProperties;
using parquet::WriterProperties;

// Data is in RBR
std::shared_ptr<arrow::RecordBatchReader> batch_stream;
ARROW_ASSIGN_OR_RAISE(batch_stream, GetRBR());

// Choose compression
std::shared_ptr<WriterProperties> props =
    WriterProperties::Builder().compression(arrow::Compression::SNAPPY)->build();

// Opt to store Arrow schema for easier reads back into Arrow
std::shared_ptr<ArrowWriterProperties> arrow_props =
    ArrowWriterProperties::Builder().store_schema()->build();

// Create a writer
std::shared_ptr<arrow::io::FileOutputStream> outfile;
ARROW_ASSIGN_OR_RAISE(outfile, arrow::io::FileOutputStream::Open(path_to_file));
std::unique_ptr<parquet::arrow::FileWriter> writer;
ARROW_ASSIGN_OR_RAISE(
    writer, parquet::arrow::FileWriter::Open(*batch_stream->schema().get(),
                                             arrow::default_memory_pool(), outfile,
                                             props, arrow_props));

// Write each batch as a row_group
for (arrow::Result<std::shared_ptr<arrow::RecordBatch>> maybe_batch : *batch_stream) {
  ARROW_ASSIGN_OR_RAISE(auto batch, maybe_batch);
  ARROW_ASSIGN_OR_RAISE(auto table,
                        arrow::Table::FromRecordBatches(batch->schema(), {batch}));
  ARROW_RETURN_NOT_OK(writer->WriteTable(*table.get(), batch->num_rows()));
}

// Write file footer and close
ARROW_RETURN_NOT_OK(writer->Close());

StreamWriter

StreamWriter 允许使用标准 C++ 输出运算符写入 Parquet 文件，类似于使用 StreamReader 类进行读取。这种类型安全的方法还确保在写入行时不会遗漏字段，并允许自动创建新的行组（在一定数据量之后）或通过使用 EndRowGroup 流修饰符显式创建新的行组。

异常用于 segnal 错误。ParquetException 在以下情况下抛出

• 尝试使用不正确的类型写入字段。

• 尝试在一行中写入太多字段。

• 尝试跳过必填字段。

#include "arrow/io/file.h"
#include "parquet/stream_writer.h"

{
   std::shared_ptr<arrow::io::FileOutputStream> outfile;

   PARQUET_ASSIGN_OR_THROW(
      outfile,
      arrow::io::FileOutputStream::Open("test.parquet"));

   parquet::WriterProperties::Builder builder;
   std::shared_ptr<parquet::schema::GroupNode> schema;

   // Set up builder with required compression type etc.
   // Define schema.
   // ...

   parquet::StreamWriter os{
      parquet::ParquetFileWriter::Open(outfile, schema, builder.build())};

   // Loop over some data structure which provides the required
   // fields to be written and write each row.
   for (const auto& a : getArticles())
   {
      os << a.name() << a.price() << a.quantity() << parquet::EndRow;
   }
}

写入器属性

要配置 Parquet 文件的写入方式，请使用 WriterProperties::Builder

#include "parquet/arrow/writer.h"
#include "arrow/util/type_fwd.h"

using parquet::WriterProperties;
using parquet::ParquetVersion;
using parquet::ParquetDataPageVersion;
using arrow::Compression;

std::shared_ptr<WriterProperties> props = WriterProperties::Builder()
   .max_row_group_length(64 * 1024)
   .created_by("My Application")
   .version(ParquetVersion::PARQUET_2_6)
   .data_page_version(ParquetDataPageVersion::V2)
   .compression(Compression::SNAPPY)
   .build();

max_row_group_length 设置每个行组的行数上限，该上限优先于写入方法中传递的 chunk_size。

您可以使用 version 设置要写入的 Parquet 版本，这将决定可以使用哪些逻辑类型。此外，您可以使用 data_page_version 设置数据页版本。默认情况下为 V1；设置为 V2 将允许更佳的压缩（跳过压缩没有空间收益的页面），但并非所有读取器都支持此数据页版本。

默认情况下压缩是关闭的，但为了充分利用 Parquet，您应该选择一种压缩编解码器。您可以为整个文件选择一个，也可以为单个列选择一个。如果您选择混合使用，则文件级选项将应用于没有特定压缩编解码器的列。选项请参阅 ::arrow::Compression。

列数据编码同样可以应用于文件级别或列级别。默认情况下，写入器将尝试对所有支持的列进行字典编码，除非字典变得过大。可以使用 disable_dictionary() 在文件级别或列级别更改此行为。当不使用字典编码时，它将回退到为列或整个文件设置的编码；默认为 Encoding::PLAIN，但这可以使用 encoding() 更改。

#include "parquet/arrow/writer.h"
#include "arrow/util/type_fwd.h"

using parquet::WriterProperties;
using arrow::Compression;
using parquet::Encoding;

std::shared_ptr<WriterProperties> props = WriterProperties::Builder()
  .compression(Compression::SNAPPY)        // Fallback
  ->compression("colA", Compression::ZSTD) // Only applies to column "colA"
  ->encoding(Encoding::BIT_PACKED)         // Fallback
  ->encoding("colB", Encoding::RLE)        // Only applies to column "colB"
  ->disable_dictionary("colB")             // Never dictionary-encode column "colB"
  ->build();

默认情况下，所有列都启用统计信息。您可以使用构建器上的 disable_statistics 禁用所有列或特定列的统计信息。有一个 max_statistics_size 限制可用于最小值和最大值的字节数，这对于字符串或二进制 blob 等类型很有用。如果列已使用 enable_write_page_index 启用页面索引，则它不会将统计信息写入页面标头，因为它在 ColumnIndex 中重复。

还有一些 Arrow 特定的设置，可以使用 parquet::ArrowWriterProperties 进行配置

#include "parquet/arrow/writer.h"

using parquet::ArrowWriterProperties;

std::shared_ptr<ArrowWriterProperties> arrow_props = ArrowWriterProperties::Builder()
   .enable_deprecated_int96_timestamps() // default False
   ->store_schema() // default False
   ->build();

这些选项主要决定 Arrow 类型如何转换为 Parquet 类型。启用 store_schema 将导致写入器将序列化的 Arrow 模式存储在文件元数据中。由于 Parquet 模式和 Arrow 模式之间没有双射，因此存储 Arrow 模式允许 Arrow 读取器更忠实地重建原始数据。这种从 Parquet 类型到原始 Arrow 类型的映射包括

• 读取带有原始时区信息的时间戳（Parquet 不支持时区）；

• 从其存储类型读取 Arrow 类型（例如从 int64 列读取 Duration）；

• 将字符串和二进制列读回具有 64 位偏移量的大型变体；

• 将列读回为字典编码的（Arrow 列和序列化的 Parquet 版本是否进行字典编码是独立的）。

支持的 Parquet 功能

Parquet 格式有许多功能，Parquet C++ 支持其中的一部分。

页面类型

页面类型	备注
DATA_PAGE
DATA_PAGE_V2
DICTIONARY_PAGE

不支持的页面类型： INDEX_PAGE。读取 Parquet 文件时，将忽略此类型的页面。

压缩

压缩编解码器	备注
SNAPPY
GZIP
BROTLI
LZ4	(1)
ZSTD

• (1) 在读取方面，Parquet C++ 能够解压缩常规 LZ4 块格式和 Parquet 参考实现 使用的 ad-hoc Hadoop LZ4 格式。在写入方面，Parquet C++ 始终生成 ad-hoc Hadoop LZ4 格式。

不支持的压缩编解码器： LZO。

编码

编码	读取	写入	备注
PLAIN	✓	✓
PLAIN_DICTIONARY	✓	✓
BIT_PACKED	✓	✓	(1)
RLE	✓	✓	(1)
RLE_DICTIONARY	✓	✓	(2)
BYTE_STREAM_SPLIT	✓	✓
DELTA_BINARY_PACKED	✓	✓
DELTA_BYTE_ARRAY	✓	✓
DELTA_LENGTH_BYTE_ARRAY	✓	✓

• (1) 仅支持编码定义和重复级别以及布尔值。

• (2) 在写入路径上，仅当在 WriterProperties::version() 中选择 Parquet 格式版本 2.4 或更高版本时，才启用 RLE_DICTIONARY。

类型

物理类型

物理类型	映射的 Arrow 类型	备注
BOOLEAN	Boolean
INT32	Int32 / 其他	(1)
INT64	Int64 / 其他	(1)
INT96	Timestamp (纳秒)	(2)
FLOAT	Float32
DOUBLE	Float64
BYTE_ARRAY	Binary / 其他	(1) (3)
FIXED_LENGTH_BYTE_ARRAY	FixedSizeBinary / 其他	(1)

• (1) 可以映射到其他 Arrow 类型，具体取决于逻辑类型（请参见下文）。

• (2) 在写入端，必须启用 ArrowWriterProperties::support_deprecated_int96_timestamps()。

• (3) 在写入端，Arrow LargeBinary 也可以映射到 BYTE_ARRAY。

逻辑类型

特定的逻辑类型可以覆盖给定物理类型的默认 Arrow 类型映射。

逻辑类型	物理类型	映射的 Arrow 类型	备注
NULL	任何	Null	(1)
INT	INT32	Int8 / UInt8 / Int16 / UInt16 / Int32 / UInt32
INT	INT64	Int64 / UInt64
DECIMAL	INT32 / INT64 / BYTE_ARRAY / FIXED_LENGTH_BYTE_ARRAY	Decimal128 / Decimal256	(2)
DATE	INT32	Date32	(3)
TIME	INT32	Time32 (毫秒)
TIME	INT64	Time64 (微秒或纳秒)
TIMESTAMP	INT64	Timestamp (毫秒、微秒或纳秒)
STRING	BYTE_ARRAY	Utf8	(4)
LIST	任何	List	(5)
MAP	任何	Map	(6)
FLOAT16	FIXED_LENGTH_BYTE_ARRAY	HalfFloat

• (1) 在写入端，将生成 Parquet 物理类型 INT32。

• (2) 在写入端，始终发出 FIXED_LENGTH_BYTE_ARRAY。

• (3) 在写入端，Arrow Date64 也映射到 Parquet DATE INT32。

• (4) 在写入端，Arrow LargeUtf8 也映射到 Parquet STRING。

• (5) 在写入端，Arrow LargeList 或 FixedSizedList 也映射到 Parquet LIST。

• (6) 在读取端，具有多个值的键不会被去重，这与 Parquet 规范 相矛盾。

不支持的逻辑类型： JSON、BSON、UUID。如果在读取 Parquet 文件时遇到此类类型，则使用默认物理类型映射（例如，Parquet JSON 列可以读取为 Arrow Binary 或 FixedSizeBinary）。

转换类型

虽然转换类型在 Parquet 格式中已弃用（它们已被逻辑类型取代），但 Parquet C++ 实现可以识别并发出它们，以便最大限度地与其他 Parquet 实现兼容。

特殊情况

Arrow 扩展类型将作为其存储类型写出。它仍然可以在读取时使用 Parquet 元数据重新创建（请参阅下面的“往返 Arrow 类型”）。

Arrow 字典类型将作为其值类型写出。它仍然可以在读取时使用 Parquet 元数据重新创建（请参阅下面的“往返 Arrow 类型”）。

往返 Arrow 类型和模式

虽然 Arrow 类型和 Parquet 类型之间没有双射，但可以将 Arrow 模式序列化为 Parquet 文件元数据的一部分。这可以使用 ArrowWriterProperties::store_schema() 启用。

在读取路径上，序列化模式将被自动识别，并将重新创建原始 Arrow 数据，并根据需要转换 Parquet 数据。

例如，将 Arrow LargeList 序列化为 Parquet 时

• 数据写出为 Parquet LIST

• 读回时，如果写入文件时启用了 ArrowWriterProperties::store_schema()，则 Parquet LIST 数据将解码为 Arrow LargeList；否则，它将解码为 Arrow List。

Parquet 字段 ID

Parquet 格式支持可选的整数*字段 ID*，可以将其分配给给定字段。例如，这在 Apache Iceberg 规范 中使用。

在写入端，如果 PARQUET:field_id 作为元数据键出现在 Arrow 字段上，则其值将被解析为非负整数，并用作相应 Parquet 字段的字段 ID。

在读取端，Arrow 会将此类字段 ID 转换为相应 Arrow 字段上名为 PARQUET:field_id 的元数据键。

序列化详细信息

Arrow 模式序列化为 Arrow IPC 模式消息，然后进行 base64 编码并存储在 Parquet 文件元数据中的 ARROW:schema 元数据键下。

限制

不支持写入或读回带有空条目的 FixedSizedList 数据。

加密

除了列索引和布隆过滤器模块的加密之外，Parquet C++ 实现了 加密规范 中指定的所有功能。

更具体地说，Parquet C++ 支持

• AES_GCM_V1 和 AES_GCM_CTR_V1 加密算法。

• 页脚、列元数据、数据页、字典页、数据页眉、字典页眉模块类型的 AAD 后缀。不支持其他模块类型（列索引、偏移索引、布隆过滤器标头、布隆过滤器位集）。

• EncryptionWithFooterKey 和 EncryptionWithColumnKey 模式。

• 加密页脚和纯文本页脚模式。

其他

特性	读取	写入	备注
列索引	✓	✓	(1)
偏移索引	✓	✓	(1)
布隆过滤器	✓	✓	(2)
CRC 校验和	✓	✓

• (1) 提供对列和偏移索引结构的访问，但数据读取 API 目前没有使用它们。

• (2) 提供了用于创建、序列化和反序列化布隆过滤器的 API，但它们未集成到数据读取 API 中。