parquet-go/parquet-go

高性能Go库，用于操作parquet文件，最初由Twilio Segment开发。

parquet-go-logo

动机

Parquet已成为在持久存储介质上表示列式数据的强大解决方案，实现了压缩和查询性能水平，使管理达到PB级的数据集成为可能。此外，密集型数据应用程序共享通用格式为我们的工具包创造了互操作的机会，为维护和操作这些系统的工程师提供了更大的杠杆作用和价值。

大规模数据管理系统的创建和演进，加上实时期望，带来了具有挑战性的维护和性能要求，现有的使用Go的parquet解决方案无法满足这些要求。

parquet-go/parquet-go包的设计和开发旨在应对这些挑战，提供高级API来读写parquet文件，同时保持低计算和内存占用，以便在数据量和成本约束要求软件达到高效率水平的环境中使用。

规范

列式存储允许Parquet比使用JSON或Protobuf更高效地存储数据。有关更多信息，请参阅Parquet格式规范。

安装

该包作为标准Go模块分发，程序可以依赖并使用以下命令安装：

go get github.com/parquet-go/parquet-go

使用该包需要Go 1.21或更高版本。

兼容性保证

该包目前作为pre-v1版本发布，这使维护人员能够自由地打破向后兼容性，以帮助改进API，因为我们了解到哪些初始设计决策需要重新审视，以更好地支持该库解决的用例。预计这些情况的发生频率很低，并且会提供文档来指导用户如何调整他们的程序以适应重大变更。

使用

以下部分描述如何使用库公开的API，通过代码示例突出显示用例，以演示它们在实践中的使用方式。

写入Parquet文件：parquet.GenericWriter[T]

parquet文件是共享相同模式的行集合，按列排列以支持对数据集子集进行更快的扫描操作。

对于简单的用例，parquet.WriteFile[T]函数允许从表示要写入文件的行的Go值切片在文件系统上创建parquet文件。

type RowType struct { FirstName, LastName string }

if err := parquet.WriteFile("file.parquet", []RowType{
    {FirstName: "Bob"},
    {FirstName: "Alice"},
}); err != nil {
    ...
}

parquet.GenericWriter[T]类型将行反规范化为列，然后将列编码为parquet文件，根据可配置的启发式方法生成行组、列块和页面。

type RowType struct { FirstName, LastName string }

writer := parquet.NewGenericWriter[RowType](output)

_, err := writer.Write([]RowType{
    ...
})
if err != nil {
    ...
}

// 关闭写入器是必要的，以刷新缓冲区并写入文件页脚。
if err := writer.Close(); err != nil {
    ...
}

在写入器上显式声明parquet模式在应用程序需要确保写入文件的数据遵循预定义模式时很有用，该模式可能与从写入器的类型参数派生的模式不同。parquet.Schema类型是parquet行模式的内存表示，从Go值的类型转换而来，可用于此目的。

schema := parquet.SchemaOf(new(RowType))
writer := parquet.NewGenericWriter[any](output, schema)
...

读取Parquet文件：parquet.GenericReader[T]

对于数据集适合内存且程序将读取文件大部分行的简单用例，parquet.ReadFile[T]函数返回表示从文件读取的行的Go值切片。

type RowType struct { FirstName, LastName string }

rows, err := parquet.ReadFile[RowType]("file.parquet")
if err != nil {
    ...
}

for _, c := range rows {
    fmt.Printf("%+v\n", c)
}

当构造读取器时，可以显式声明行的预期模式，这对于确保程序接收到匹配特定格式的行很有用；例如，在处理来自应用程序无法信任使用预期模式的远程存储源的文件时。

通过在构造读取器时将parquet.Schema实例作为参数传递来配置读取器的模式。当声明模式时，包实现的转换规则将被应用，以确保应用程序读取的行匹配所需格式（参见演进Parquet模式）。

schema := parquet.SchemaOf(new(RowType))
reader := parquet.NewReader(file, schema)
...

检查Parquet文件：parquet.File

有时，低级API对于利用parquet文件的列式布局很有用。parquet.File类型旨在为Go应用程序提供此类功能，通过公开API来迭代parquet文件的各个部分。

f, err := parquet.OpenFile(file, size)
if err != nil {
    ...
}

for _, rowGroup := range f.RowGroups() {
    for _, columnChunk := range rowGroup.ColumnChunks() {
        ...
    }
}

演进Parquet模式：parquet.Convert

Parquet文件嵌入了解释其内容所需的所有元数据，包括对它们包含的行和列所表示的表的模式的描述。

Parquet文件也是不可变的；一旦写入，就没有更新文件的机制。如果需要更改其内容，必须读取行，修改，然后写入新文件。

由于应用程序在演进，写入parquet文件的模式也倾向于随时间演进。当应用程序需要对具有异构模式的parquet文件进行操作时，这些需求会产生挑战：期望存在新列的算法在处理来自具有不匹配模式版本的文件的行时可能会遇到问题。

为了帮助构建能够处理演进模式的应用程序，parquet-go/parquet-go实现了转换规则，创建行组的视图以在模式版本之间进行转换。

parquet.Convert函数是从源模式到目标模式构造转换规则的低级例程。该函数用于构建parquet.RowReader或parquet.RowGroup的转换视图，例如：

type RowTypeV1 struct { ID int64; FirstName string }
type RowTypeV2 struct { ID int64; FirstName, LastName string }

source := parquet.SchemaOf(RowTypeV1{})
target := parquet.SchemaOf(RowTypeV2{})

conversion, err := parquet.Convert(target, source)
if err != nil {
    ...
}

targetRowGroup := parquet.ConvertRowGroup(sourceRowGroup, conversion)
...

当传递给函数的读取器和写入器还实现parquet.RowReaderWithSchema和parquet.RowWriterWithSchema接口时，parquet.CopyRows函数会自动应用转换规则。复制确定读取器和写入器模式是否可以相互转换，并自动应用转换规则以促进模式之间的转换。

目前，转换规则仅支持从模式中添加或删除列，不执行类型转换，也没有重命名列等方式。未来可能会添加更高级的转换规则。

排序行组：parquet.GenericBuffer[T]

parquet.GenericWriter[T]类型针对最小内存使用进行了优化，保持行的顺序不变，并在页面填满后立即刷新。

Parquet支持通过在行组上声明排序列来表示行的排序列。排序行组需要在排序和写入parquet文件之前缓冲所有行。

为了帮助这些用例，parquet-go/parquet-go包公开了parquet.GenericBuffer[T]类型，它充当行的缓冲区，并实现sort.Interface以允许应用程序在将行写入文件之前对其进行排序。

行排序的列在创建parquet.GenericBuffer[T]实例时使用parquet.SortingColumns函数配置，该函数构造配置缓冲区的行组选项。parquet列的类型定义了如何比较值，详见Parquet逻辑类型。

写入文件时，缓冲区被具体化为具有声明的排序列的单个行组。写入后，可以通过调用其Reset方法重用缓冲区。

以下示例展示了如何使用parquet.GenericBuffer[T]对写入parquet文件的行进行排序：

type RowType struct { FirstName, LastName string }

buffer := parquet.NewGenericBuffer[RowType](
    parquet.SortingRowGroupConfig(
        parquet.SortingColumns(
            parquet.Ascending("LastName"),
            parquet.Ascending("FistName"),
        ),
    ),
)

buffer.Write([]RowType{
    {FirstName: "Luke", LastName: "Skywalker"},
    {FirstName: "Han", LastName: "Solo"},
    {FirstName: "Anakin", LastName: "Skywalker"},
})

sort.Sort(buffer)

writer := parquet.NewGenericWriter[RowType](output)
_, err := parquet.CopyRows(writer, buffer.Rows())
if err != nil {
    ...
}
if err := writer.Close(); err != nil {
    ...
}

合并行组：parquet.MergeRowGroups

Parquet文件通常作为数据处理或存储层底层引擎的一部分使用，在这种情况下，将多个行组合并为一个包含更多行的行组可能是提高查询性能的有用操作；例如，parquet文件中的布隆过滤器为每个行组存储，行组越大，需要存储的过滤器就越少，它们就变得越有效。

parquet-go/parquet-go包支持创建行组的合并视图，其中视图包含合并组的所有行，保持由组的排序列定义的顺序。

合并行组时有一些限制：

所有行组的排序列必须相同，或者合并操作必须明确配置一组排序列，这些排序列是所有合并行组的排序列的前缀。
行组的模式必须全部相等，或者合并操作必须明确配置一个所有行组都可以转换到的模式，在这种情况下，模式转换的限制适用。

一旦创建了合并视图，可以将其写入新的parquet文件或缓冲区，以创建更大的行组：

merge, err := parquet.MergeRowGroups(rowGroups)
if err != nil {
    ...
}

writer := parquet.NewGenericWriter[RowType](output)
_, err := parquet.CopyRows(writer, merge)
if err != nil {
    ...
}
if err := writer.Close(); err != nil {
    ...
}

使用布隆过滤器：parquet.BloomFilter

Parquet文件可以嵌入布隆过滤器，以帮助提高文件中点查找的性能。parquet布隆过滤器的格式在parquet规范中有文档说明：Parquet布隆过滤器

默认情况下，parquet文件中不创建布隆过滤器，但应用程序可以在实例化写入器时使用parquet.BloomFilters选项配置要创建过滤器的列列表；例如：

type RowType struct {
    FirstName string `parquet:"first_name"`
    LastName  string `parquet:"last_name"`
}

const filterBitsPerValue = 10
writer := parquet.NewGenericWriter[RowType](output,
    parquet.BloomFilters(
        // 
用于读取 parquet 文件的低级 API 提供了更高效的方式来访问列值。连续的值序列被分组到页面中,由 `parquet.Page` 接口表示。

一个列块可能包含多个页面,每个页面持有一部分列值。应用程序可以通过将它们读入 `parquet.Value` 缓冲区,或者对页面进行类型断言以读取原始 Go 值数组来检索列值。以下示例演示了如何使用这两种机制来读取列值:

```go
pages := column.Pages()
defer func() {
    checkErr(pages.Close())
}()

for {
    p, err := pages.ReadPage()
    if err != nil {
        ... // 当没有更多页面时为 io.EOF
    }

    switch page := p.Values().(type) {
    case parquet.Int32Reader:
        values := make([]int32, page.NumValues())
        _, err := page.ReadInt32s(values)
        ...
    case parquet.Int64Reader:
        values := make([]int64, page.NumValues())
        _, err := page.ReadInt64s(values)
        ...
    default:
        values := make([]parquet.Value, page.NumValues())
        _, err := page.ReadValues(values)
        ...
    }
}

在对值进行聚合时,读取类型化值数组通常更可取,因为这种模型在内存中提供了更紧凑的值表示,并且与 SIMD 矢量化等优化的使用很好地配合。

优化写入

处理列式存储的应用程序有时被设计为在整个抽象层中使用列式数据;然后就可以直接写入值列,而不是从列值重构行。该包提供了两种主要机制来满足这些用例:

A. 写入类型化数组列

第一种解决方案假设程序使用内存中的类型化值数组,例如 []float32 之类的原始 Go 类型切片;如果应用程序构建在 Apache Arrow 等框架之上,就会出现这种情况。

parquet.GenericBuffer[T] 是 parquet.RowGroup 接口的一个实现,它维护列值的内存缓冲区。可以通过将原始值装箱到 parquet.Value 数组中,或者对列进行类型断言以访问接受 Go 原始类型数组的专用版本写入方法来写入行。

使用这两种模型时,应用程序负责确保写入每列的行数相同,否则生成的 parquet 文件将会格式错误。

以下示例演示了如何使用这两种模型来写入 Go 值列:

type RowType struct { FirstName, LastName string }

func writeColumns(buffer *parquet.GenericBuffer[RowType], firstNames []string) error {
    values := make([]parquet.Value, len(firstNames))
    for i := range firstNames {
        values[i] = parquet.ValueOf(firstNames[i])
    }
    _, err := buffer.ColumnBuffers()[0].WriteValues(values)
    return err
}

type RowType struct { ID int64; Value float32 }

func writeColumns(buffer *parquet.GenericBuffer[RowType], ids []int64, values []float32) error {
    if len(ids) != len(values) {
        return fmt.Errorf("id 和值的数量不匹配: ids=%d values=%d", len(ids), len(values))
    }
    columns := buffer.ColumnBuffers()
    if err := columns[0].(parquet.Int64Writer).WriteInt64s(ids); err != nil {
        return err
    }
    if err := columns[1].(parquet.FloatWriter).WriteFloats(values); err != nil {
        return err
    }
    return nil
}

后者更高效,因为它不需要将输入装箱到 parquet.Value 的中间数组中。但是,根据情况不同,它可能并不总是正确的模型,有时通用抽象可能是更具表现力的模型。

B. 实现 parquet.RowGroup

需要完全控制行组构建的程序可以选择提供自己的 parquet.RowGroup 接口实现,其中包括定义 parquet.ColumnChunk 和 parquet.Page 的实现以暴露行组的列值。

当需要进一步优化底层存储或数据的内存表示,超出使用 parquet.GenericBuffer[T] 的中间缓冲层所能实现的程度时,这种模型可能更可取。

有关完整的接口文档,请参阅 parquet.RowGroup。

C. 使用磁盘页面缓冲区

在生成 parquet 文件时,写入器需要在创建行组之前缓冲所有页面。这可能需要大量内存,因为在生成之前必须缓冲整个文件内容。在某些情况下,文件甚至可能大于程序可用的内存量。

parquet.GenericWriter[T] 可以配置为在生成文件时使用磁盘存储作为临时缓冲区,方法是使用 parquet.ColumnPageBuffers 选项和 parquet.PageBufferPool 接口配置不同的页面缓冲池。

parquet-go/parquet-go 包提供了该接口的一个实现,它使用临时文件来存储生成文件时的页面,允许程序使用本地存储作为交换空间来保存页面并将内存使用保持在最低水平。以下示例演示了如何配置 parquet 写入器以使用磁盘页面缓冲区:

type RowType struct { ... }

writer := parquet.NewGenericWriter[RowType](output,
    parquet.ColumnPageBuffers(
        parquet.NewFileBufferPool("", "buffers.*"),
    ),
)

当行组完成时,需要将缓冲到磁盘的页面复制回输出文件。这会导致 I/O 操作和存储空间需求加倍(系统需要有足够的可用磁盘空间来保存文件的两个副本)。如果文件系统支持磁盘页面的写时复制,由于 os.File 实例之间的复制使用 copy_file_range(2) (在 linux 上)进行了优化,内核通常可以优化掉由此产生的写入放大。

有关完整的接口文档,请参阅 parquet.PageBufferPool。