读取和写入 Parquet 文件#

另请参见

The Parquet format 是用于复杂数据的空间效率高的列式存储格式。Parquet C++ 实现是 Apache Arrow 项目的一部分，并得益于与 Arrow C++ 类和功能的紧密集成。

读取 Parquet 文件#

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

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

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

FileReader#

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

// #include "arrow/io/api.h"
// #include "arrow/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_RETURN_NOT_OK(parquet::arrow::OpenFile(input, pool, &arrow_reader));

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

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

不要在 parquet::ArrowReaderProperties 中打开读取合并（预缓冲）。
使用 arrow::FileReader::GetRecordBatchReader() 按批次读取数据。
在 parquet::ReaderProperties 中打开 enable_buffered_stream。

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

StreamReader#

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

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

异常用于信号错误。一个 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#

The 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#

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

异常用于信号错误。一个 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();

The 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 实现使用的特定 Hadoop LZ4 格式。在写入方面，Parquet C++ 始终生成特定 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	布尔值
INT32	Int32 / 其他	(1)
INT64	Int64 / 其他	(1)
INT96	时间戳（纳秒）	(2)
FLOAT	Float32
DOUBLE	Float64
BYTE_ARRAY	二进制 / 其他	(1) (3)
FIXED_LENGTH_BYTE_ARRAY	固定大小二进制 / 其他	(1)

(1) 可以映射到其他 Arrow 类型，具体取决于逻辑类型（见下文）。
(2) 在写入方面，必须启用 ArrowWriterProperties::support_deprecated_int96_timestamps()。
(3) 在写入方面，Arrow 大二进制也可以映射到 BYTE_ARRAY。

逻辑类型#

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

逻辑类型	物理类型	映射的 Arrow 类型	笔记
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	时间戳（毫秒、微秒或纳秒）
STRING	BYTE_ARRAY	Utf8	(4)
LIST	任何	列表	(5)
MAP	任何	映射	(6)
FLOAT16	FIXED_LENGTH_BYTE_ARRAY	半精度浮点数

(1) 在写入方面，会生成 Parquet 物理类型 INT32。
(2) 在写入方面，始终会发出 FIXED_LENGTH_BYTE_ARRAY。
(3) 在写入方面，Arrow Date64 也映射到 Parquet DATE INT32。
(4) 在写入方面，Arrow 大 Utf8 也映射到 Parquet STRING。
(5) 在写入方面，Arrow 大列表或固定大小列表也映射到 Parquet LIST。
(6) 在读取方面，具有多个值的键不会被去重，这与 Parquet 规范相矛盾。

不支持的逻辑类型：JSON、BSON、UUID。如果在读取 Parquet 文件时遇到此类类型，则使用默认物理类型映射（例如，Parquet JSON 列可能被读取为 Arrow 二进制或固定大小二进制）。

转换类型#

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

特殊情况#

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

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

Arrow 类型和模式的往返#

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

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

例如，在将 Arrow 大列表序列化到 Parquet 时

数据将作为 Parquet LIST 写出
读回时，如果在写入文件时启用了 ArrowWriterProperties::store_schema()，则 Parquet LIST 数据将被解码为 Arrow 大列表；否则，它将被解码为 Arrow 列表。

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 元数据键下。

限制#

不支持写入或读回具有空条目的固定大小列表数据。

加密#

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

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

AES_GCM_V1 和 AES_GCM_CTR_V1 加密算法。
用于页脚、列元数据、数据页、字典页、数据页页眉、字典页页眉模块类型的 AAD 后缀。其他模块类型（列索引、偏移量索引、布隆过滤器页眉、布隆过滤器位集）不受支持。
EncryptionWithFooterKey 和 EncryptionWithColumnKey 模式。
Encrypted Footer 和 Plaintext Footer 模式。

杂项#

功能	读取	写入	笔记
列索引	✓	✓	(1)
偏移量索引	✓	✓	(1)
布隆过滤器	✓	✓	(2)
CRC 校验和	✓	✓

(1) 提供了对列和偏移量索引结构的访问，但数据读取 API 目前没有使用它们。
(2) 提供了用于创建、序列化和反序列化布隆过滤器的 API，但它们没有集成到数据读取 API 中。