Go语言处理Parquet文件：写入、分页读取与列计算实战

洛溪

发布时间：2026-10-13| 4人浏览过

原创

本文详解Golang使用parquet-go库操作Parquet文件的完整流程。从结构体字段映射与标签定义入手，逐步讲解数据写入、文件压缩配置以及针对大文件的内存优化分页读取技术，并提供基于列路径的聚合计算示例，帮助开发者在大数据场景下高效处理二进制列存储格式。

理解结构与Tag定义

Parquet是一种面向列存储的二进制文件格式，广泛应用于Hadoop生态及数据科学环境中。在Go中处理此类文件时，需利用结构体（Struct）映射数据行，并通过struct tag控制各字段在Parquet中的表现。

使用 github.com/xitongsys/parquet-go 包时，tag的核心作用是指定字段名、数据类型及编码方式。例如：

type user struct {
  ID        string    `parquet:"name=id, type=BYTE_ARRAY, encoding=PLAIN_DICTIONARY"`
  FirstName string    `parquet:"name=firstname, type=BYTE_ARRAY, encoding=PLAIN_DICTIONARY"`
  LastName  string    `parquet:"name=lastname, type=BYTE_ARRAY, encoding=PLAIN_DICTIONARY"`
  Email     string    `parquet:"name=email, type=BYTE_ARRAY, encoding=PLAIN_DICTIONARY"`
  Phone     string    `parquet:"name=phone, type=BYTE_ARRAY, encoding=PLAIN_DICTIONARY"`
  Blog      string    `parquet:"name=blog, type=BYTE_ARRAY, encoding=PLAIN_DICTIONARY"`
  Username  string    `parquet:"name=username, type=BYTE_ARRAY, encoding=PLAIN_DICTIONARY"`
  Score     float64   `parquet:"name=score, type=DOUBLE"`
  CreatedAt time.Time //该字段未指定tag，不会保存至Parquet文件中
}

注意Go类型与Parquet原生类型的对应关系，如 float64 对应 DOUBLE，字符串切片对应 BYTE_ARRAY。未被标记的字段（如时间戳字段若未加tag）将默认被忽略。

第1步：生成Parquet文件并写入数据

目标：将内存中的数据对象持久化为列存储的 .parquet 文件，并通过设置压缩算法减小体积。

入口：初始化本地文件写入器与Parquet写入器。
动作：通过循环调用Write方法逐行写入，最后停止写入流。
关键参数：CompressionType = parquet.CompressionCodec_GZIP 用于配置GZIP压缩。
完成状态：磁盘上生成包含数据的输出文件。

package main

import (
  "fmt"
  "log"
  "time"
  "github.com/bxcodec/faker/v3"
  "github.com/xitongsys/parquet-go-source/local"
  "github.com/xitongsys/parquet-go/parquet"
  "github.com/xitongsys/parquet-go/writer"
)

const recordNumber = 10000

func main() {
  var data []*user
  // 创建模拟测试数据
  for i := 0; i < recordNumber; i++ {
    u := &user{
      ID:        faker.UUIDDigit(),
      FirstName: faker.FirstName(),
      LastName:  faker.LastName(),
      Email:     faker.Email(),
      Phone:     faker.Phonenumber(),
      Blog:      faker.URL(),
      Username:  faker.Username(),
      Score:     float64(i),
      CreatedAt: time.Now(),
    }
    data = append(data, u)
  }

  err := generateParquet(data)
  if err != nil {
    log.Fatal(err)
  }
}

func generateParquet(data []*user) error {
  log.Println("generating parquet file")
  // 1. 建立本地文件输出通道
  fw, err := local.NewLocalFileWriter("output.parquet")
  if err != nil {
    return err
  }
  defer fw.Close()

  // 2. 构建Parquet写 writer，需传入结构体类型及预估行数
  pw, err := writer.NewParquetWriter(fw, new(user), int64(len(data)))
  if err != nil {
    return err
  }

  // 3. 设置压缩方式为 GZIP
  pw.CompressionType = parquet.CompressionCodec_GZIP

  // 4. 循环写入数据行
  for _, d := range data {
    if err = pw.Write(d); err != nil {
      return err
    }
  }

  // 5. 结束写入，确保文件头尾信息正确闭合
  if err = pw.WriteStop(); err != nil {
    return err
  }
  return nil
}

第2步：读取文件（一次性与分页模式对比）

Parquet是列式存储，解压后体积可能巨大。官方建议对于大文件不要一次性全量读入内存，以避免发生 OOM（内存溢出）。

小文件一次性读取

适用于数据量极小的场景，直接加载所有记录：

func readParquet() ([]*user, error) {
  fr, err := local.NewLocalFileReader("output.parquet")
  if err != nil {
    return nil, err
  }
  pr, err := reader.NewParquetReader(fr, new(user), recordNumber)
  if err != nil {
    return nil, err
  }

  u := make([]*user, recordNumber)
  // 将所有数据读入切片
  if err = pr.Read(&u); err != nil {
    return nil, err
  }
  pr.ReadStop()
  fr.Close()
  return u, nil
}

大数据量分页读取（推荐）

通过 pr.SkipRows() 跳过已处理的行数，实现类似“游标”的分页效果，大幅降低内存消耗。

func readPartialParquet(pageSize, page int) ([]*user, error) {
  fr, err := local.NewLocalFileReader("output.parquet")
  if err != nil {
    return nil, err
  }
  defer func() { _ = fr.Close() }()

  // 每次只申请 pageSize 大小的 Reader
  pr, err := reader.NewParquetReader(fr, new(user), int64(pageSize))
  if err != nil {
    return nil, err
  }
  defer pr.ReadStop()

  // 核心：跳过前 (page * pageSize) 条数据
  pr.SkipRows(int64(pageSize * page))

  u := make([]*user, pageSize)
  if err = pr.Read(&u); err != nil {
    return nil, err
  }

  return u, nil
}

结合 pr.GetNumRows() 获取总行数，即可推算出总页数进行遍历。

第3步：计算特定列的平均值

利用Parquet列存特性，不读取整行仅读取特定列数据，可大幅提升统计效率。此处演示计算 Score 字段的平均值。

func calcScoreAVG() (float64, error) {
  fr, err := local.NewLocalFileReader("output.parquet")
  if err != nil {
    return 0.0, err
  }
  // 使用 ColumnReader 专门针对列读取
  pr, err := reader.NewParquetColumnReader(fr, recordNumber)
  if err != nil {
    return 0.0, err
  }
  num := int(pr.GetNumRows())

  // 获取列数据，路径由 root + \u0001分隔符 + 字段名组成
  data, _, _, err := pr.ReadColumnByPath("parquet_go_root\u0001score", num)
  if err != nil {
    return 0.0, err
  }

  var result float64
  for _, val := range data {
    result += val.(float64)
  }
  return (result / float64(num)), nil
}