如何使用Golang Apache Arrow库读取Parquet的重复字段？

在Arrow Go中，Parquet的重复字段会被映射为ListArray类型（对应Arrow的list逻辑类型）。以下是具体的实现步骤和代码示例：

核心步骤 #

• 初始化Parquet文件读取器，绑定Arrow的Schema
• 逐批次读取Parquet数据，将目标列转换为ListArray
• 遍历ListArray，通过偏移量（offset）和长度（length）提取每个行的重复元素集合

完整代码示例 #

假设你的Parquet文件包含一个名为tags的重复字符串字段，以下是读取逻辑：

package main

import (
	"fmt"
	"os"

	"github.com/apache/arrow/go/v15/arrow"
	"github.com/apache/arrow/go/v15/arrow/array"
	"github.com/apache/arrow/go/v15/parquet/pqarrow"
)

func main() {
	// 1. 打开Parquet文件
	file, err := os.Open("data.parquet")
	if err != nil {
		panic(err)
	}
	defer file.Close()

	// 2. 创建Parquet读取器
	reader, err := pqarrow.NewFileReader(file, pqarrow.ArrowReadProperties{})
	if err != nil {
		panic(err)
	}
	defer reader.Close()

	// 3. 获取文件Schema，确认重复字段的类型
	schema := reader.Schema()
	tagsFieldIdx := schema.FieldIdx("tags")
	if tagsFieldIdx == -1 {
		panic("field 'tags' not found")
	}
	tagsField := schema.Field(tagsFieldIdx)
	// 验证字段是否为重复类型（List）
	if tagsField.Type.ID() != arrow.List {
		panic("'tags' is not a repeated field")
	}

	// 4. 逐批次读取数据
	batchReader, err := reader.GetRecordReader()
	if err != nil {
		panic(err)
	}
	defer batchReader.Release()

	for batchReader.Next() {
		record := batchReader.Record()
		defer record.Release()

		// 将列转换为ListArray
		tagsCol := record.Column(tagsFieldIdx).(*array.List)
		// 获取ListArray的子数组（即重复元素的实际存储）
		tagsValues := tagsCol.ListValues().(*array.String)

		// 遍历每一行的重复字段
		for i := 0; i < tagsCol.Len(); i++ {
			if tagsCol.IsNull(i) {
				fmt.Printf("Row %d: null\n", i)
				continue
			}
			// 获取当前行重复元素的偏移和长度
			offset := tagsCol.Offset(i)
			length := tagsCol.Length(i)
			// 提取对应范围的元素
			rowTags := make([]string, length)
			for j := 0; j < length; j++ {
				rowTags[j] = tagsValues.Value(offset + j)
			}
			fmt.Printf("Row %d: %v\n", i, rowTags)
		}
	}

	if err := batchReader.Err(); err != nil {
		panic(err)
	}
}

关键细节说明 #

• ListArray通过Offset(i)和Length(i)确定第i行重复元素在子数组中的位置范围
• 子数组的类型由重复字段的实际类型决定（示例中是StringArray，如果是整数则为Int64Array等）
• 注意处理IsNull(i)的情况，避免空指针异常

内容的提问来源于stack exchange，提问作者flexwang