使用parquet-go高效读写Parquet文件

Parquet是一种列式存储格式，在大数据处理中非常流行。在Go语言中，parquet-go是一个高效的Parquet文件读写库。下面我将详细介绍如何使用这个库。

安装

首先安装parquet-go库：

go get github.com/xitongsys/parquet-go

基本使用

1. 定义结构体

Parquet文件需要与结构体映射，我们先定义一个结构体：

type Student struct {
    Name    string  `parquet:"name=name, type=UTF8"`
    Age     int32   `parquet:"name=age, type=INT32"`
    Id      int64   `parquet:"name=id, type=INT64"`
    Weight  float32 `parquet:"name=weight, type=FLOAT"`
    Sex     bool    `parquet:"name=sex, type=BOOLEAN"`
    Day     int32   `parquet:"name=day, type=DATE"`
}

2. 写入Parquet文件

import (
    "github.com/xitongsys/parquet-go/parquet"
    "github.com/xitongsys/parquet-go/writer"
    "os"
)

func writeParquet() error {
    // 创建文件
    fw, err := os.Create("output.parquet")
    if err != nil {
        return err
    }
    defer fw.Close()

    // 创建Parquet writer
    pw, err := writer.NewParquetWriter(fw, new(Student), 4)
    if err != nil {
        return err
    }

    // 设置压缩方式
    pw.CompressionType = parquet.CompressionCodec_SNAPPY

    // 写入数据
    students := []Student{
        {Name: "张三", Age: 23, Id: 1, Weight: 60.5, Sex: true, Day: 18687},
        {Name: "李四", Age: 24, Id: 2, Weight: 70.5, Sex: false, Day: 18688},
    }

    for _, stu := range students {
        if err = pw.Write(stu); err != nil {
            return err
        }
    }

    // 关闭writer
    if err = pw.WriteStop(); err != nil {
        return err
    }

    return nil
}

3. 读取Parquet文件

import (
    "github.com/xitongsys/parquet-go/reader"
    "os"
    "log"
)

func readParquet() ([]Student, error) {
    fr, err := os.Open("output.parquet")
    if err != nil {
        return nil, err
    }
    defer fr.Close()

    pr, err := reader.NewParquetReader(fr, new(Student), 4)
    if err != nil {
        return nil, err
    }
    defer pr.ReadStop()

    num := int(pr.GetNumRows())
    students := make([]Student, num)
    if err = pr.Read(&students); err != nil {
        return nil, err
    }

    return students, nil
}

func main() {
    if err := writeParquet(); err != nil {
        log.Fatal(err)
    }
    
    students, err := readParquet()
    if err != nil {
        log.Fatal(err)
    }
    
    for _, stu := range students {
        log.Printf("%+v", stu)
    }
}

高级特性

1. 处理嵌套结构

type Address struct {
    City string `parquet:"name=city, type=UTF8"`
    Zip  string `parquet:"name=zip, type=UTF8"`
}

type Person struct {
    Name    string  `parquet:"name=name, type=UTF8"`
    Age     int32   `parquet:"name=age, type=INT32"`
    Addr    Address `parquet:"name=addr"`
}

2. 使用内存缓冲区

func writeToBuffer() ([]byte, error) {
    buf := new(bytes.Buffer)
    pw, err := writer.NewParquetWriter(buf, new(Student), 4)
    if err != nil {
        return nil, err
    }
    
    // ... 写入数据 ...
    
    if err = pw.WriteStop(); err != nil {
        return nil, err
    }
    
    return buf.Bytes(), nil
}

3. 性能优化建议

1. 批量写入数据而不是单条写入
2. 选择合适的压缩算法(SNAPPY通常是不错的选择)
3. 调整RowGroup大小(默认128MB)
4. 对于大文件，考虑并行处理

错误处理

func safeRead() {
    students, err := readParquet()
    if err != nil {
        if pe, ok := err.(*parquet.ParquetError); ok {
            log.Printf("Parquet error: %s, code: %d", pe.Message, pe.Code)
        } else {
            log.Printf("Other error: %v", err)
        }
        return
    }
    // 处理数据...
}

parquet-go库提供了高效的Parquet文件处理能力，支持各种数据类型和嵌套结构。通过合理使用，可以在Go应用中高效地处理列式存储数据。

对于更复杂的场景，还可以考虑使用parquet-go提供的低级API直接操作列数据，但这需要更深入的理解Parquet文件格式。