Golang處理parquet文件實(shí)戰(zhàn)指南

更新時(shí)間：2023年03月07日 09:17:06 作者：夢(mèng)想畫(huà)家

這篇文章主要給大家介紹了關(guān)于Golang處理parquet文件的相關(guān)資料,文中通過(guò)實(shí)例代碼介紹的非常詳細(xì),對(duì)大家學(xué)習(xí)或者使用Golang具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

前言

Parquet是Apache基金會(huì)支持的項(xiàng)目，是面向列存儲(chǔ)二進(jìn)制文件格式。支持不同類(lèi)型的壓縮方式，廣泛用于數(shù)據(jù)科學(xué)和大數(shù)據(jù)環(huán)境，如Hadoop生態(tài)。

本文主要介紹Go如何生成和處理parquet文件。

創(chuàng)建結(jié)構(gòu)體

首先創(chuàng)建struct，用于表示要處理的數(shù)據(jù)：

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 //wont be saved in the parquet file
}

這里要提醒的是tag,用于說(shuō)明struct中每個(gè)字段在生成parquet過(guò)程中如何被處理。

parquet-go包可以處理parquet數(shù)據(jù)，更多的tag可以參考其官網(wǎng)。

生成parquet文件

下面現(xiàn)給出生成parquet文件的代碼，然后分別進(jìn)行說(shuō)明：

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/reader"
  "github.com/xitongsys/parquet-go/writer"
)

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 //wont be saved in the parquet file
}

const recordNumber = 10000

func main() {
  var data []*user
  //create fake data
  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")
  fw, err := local.NewLocalFileWriter("output.parquet")
  if err != nil {
    return err
  }
  //parameters: writer, type of struct, size
  pw, err := writer.NewParquetWriter(fw, new(user), int64(len(data)))
  if err != nil {
    return err
  }
  //compression type
  pw.CompressionType = parquet.CompressionCodec_GZIP
  defer fw.Close()
  for _, d := range data {
    if err = pw.Write(d); err != nil {
      return err
    }
  }
  if err = pw.WriteStop(); err != nil {
    return err
  }
  return nil
}

定義結(jié)構(gòu)體上面已經(jīng)說(shuō)明，但需要提醒的是類(lèi)型與文檔保持一致：

Primitive Type	Go Type
BOOLEAN	bool
INT32	int32
INT64	int64
INT96(deprecated)	string
FLOAT	float32
DOUBLE	float64
BYTE_ARRAY	string
FIXED_LEN_BYTE_ARRAY	string

接著就是使用faker包生成模擬數(shù)據(jù)。然后調(diào)用err := generateParquet(data)方法。該方法大概邏輯為：

首先準(zhǔn)備輸出文件，然后基于本地輸出文件構(gòu)造pw,用于寫(xiě)parquet數(shù)據(jù)：

  fw, err := local.NewLocalFileWriter("output.parquet")
  if err != nil {
    return err
  }
  //parameters: writer, type of struct, size
  pw, err := writer.NewParquetWriter(fw, new(user), int64(len(data)))
  if err != nil {
    return err
  }

  //compression type
  pw.CompressionType = parquet.CompressionCodec_GZIP
  defer fw.Close()

然后設(shè)置壓縮類(lèi)型，并通過(guò)defer操作確保關(guān)閉文件。下面開(kāi)始寫(xiě)數(shù)據(jù)：

  for _, d := range data {
    if err = pw.Write(d); err != nil {
      return err
    }
  }
  if err = pw.WriteStop(); err != nil {
    return err
  }
  return nil

循環(huán)寫(xiě)數(shù)據(jù)，最后調(diào)用pw.WriteStop()停止寫(xiě)。成功寫(xiě)文件后，下面介紹如何讀取parquet文件。

讀取parquet文件

首先介紹如何一次性讀取文件，主要用于讀取較小的文件：

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
}

大概流程如下：首先定義本地文件，然后構(gòu)造pr用于讀取parquet文件：

  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
  }

然后定義目標(biāo)內(nèi)容容器u，一次性讀取數(shù)據(jù)：

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

但一次性大量記錄加載至內(nèi)存可能有問(wèn)題。這是官方文檔提示：

If the parquet file is very big (even the size of parquet file is small, the uncompressed size may be very large), please don’t read all rows at one time, which may induce the OOM. You can read a small portion of the data at a time like a stream-oriented file.

大意是不要一次讀取文件至內(nèi)存，可能造成OOM。實(shí)際應(yīng)用中應(yīng)該分頁(yè)讀取，下面通過(guò)代碼進(jìn)行說(shuō)明：

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

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

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

	return u, nil
}

與上面函數(shù)差異不大，首先函數(shù)包括兩個(gè)參數(shù)，用于指定頁(yè)大小和頁(yè)數(shù)，關(guān)鍵代碼是跳過(guò)一定記錄：

  pr.SkipRows(int64(pageSize * page))

根據(jù)這個(gè)方法可以獲得總行數(shù)，pr.GetNumRows()，然后結(jié)合頁(yè)大小計(jì)算總頁(yè)數(shù)，最后循環(huán)可以實(shí)現(xiàn)分頁(yè)查詢(xún)。

計(jì)算列平均值

既然使用了Parquet列存儲(chǔ)格式，下面演示下如何計(jì)算Score列的平均值。

func calcScoreAVG() (float64, error) {
  fr, err := local.NewLocalFileReader("output.parquet")
  if err != nil {
    return 0.0, err
  }
  pr, err := reader.NewParquetColumnReader(fr, recordNumber)
  if err != nil {
    return 0.0, err
  }
  num := int(pr.GetNumRows())

  data, _, _, err := pr.ReadColumnByPath("parquet_go_root\u0001score", num)
  if err != nil {
    return 0.0, err
  }
  var result float64
  for _, i := range data {
    result += i.(float64)
  }
  return (result / float64(num)), nil
}

首先打開(kāi)文件，然后調(diào)用pr.GetNumRows()方法獲取總行數(shù)。然后基于路徑指定列，其中parquet_go_root為根路徑，因?yàn)榍懊媸褂米止?jié)數(shù)組，這里分割符變?yōu)閈u0001，完整路徑為：parquet_go_root\u0001score。