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

更新時間：2023年03月07日 09:17:06 作者：夢想畫家

這篇文章主要給大家介紹了關于Golang處理parquet文件的相關資料,文中通過實例代碼介紹的非常詳細,對大家學習或者使用Golang具有一定的參考學習價值,需要的朋友可以參考下

前言

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

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

創(chuàng)建結構體

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

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,用于說明struct中每個字段在生成parquet過程中如何被處理。

parquet-go包可以處理parquet數據，更多的tag可以參考其官網。

生成parquet文件

下面現給出生成parquet文件的代碼，然后分別進行說明：

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
}

定義結構體上面已經說明，但需要提醒的是類型與文檔保持一致：

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包生成模擬數據。然后調用err := generateParquet(data)方法。該方法大概邏輯為：

首先準備輸出文件，然后基于本地輸出文件構造pw,用于寫parquet數據：

  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()

然后設置壓縮類型，并通過defer操作確保關閉文件。下面開始寫數據：

  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)寫數據，最后調用pw.WriteStop()停止寫。成功寫文件后，下面介紹如何讀取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
}

大概流程如下：首先定義本地文件，然后構造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
  }

然后定義目標內容容器u，一次性讀取數據：

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

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

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.

大意是不要一次讀取文件至內存，可能造成OOM。實際應用中應該分頁讀取，下面通過代碼進行說明：

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
}

與上面函數差異不大，首先函數包括兩個參數，用于指定頁大小和頁數，關鍵代碼是跳過一定記錄：

  pr.SkipRows(int64(pageSize * page))

根據這個方法可以獲得總行數，pr.GetNumRows()，然后結合頁大小計算總頁數，最后循環(huán)可以實現分頁查詢。

計算列平均值

既然使用了Parquet列存儲格式，下面演示下如何計算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
}

首先打開文件，然后調用pr.GetNumRows()方法獲取總行數。然后基于路徑指定列，其中parquet_go_root為根路徑，因為前面使用字節(jié)數組，這里分割符變?yōu)閈u0001，完整路徑為：parquet_go_root\u0001score。

總結

到此這篇關于Golang處理parquet文件的文章就介紹到這了,更多相關Golang處理parquet文件內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

簡單對比一下?C語言?與?Go語言
這篇文章主要介紹了簡單對比一下?C語言?與?Go語言的相關資料,需要的朋友可以參考下
2023-08-08
Golang中的變量學習小結
本文主要帶大家學習了Golang里面的四大類型的變量，十分的詳細，有需要的小伙伴可以參考下
2018-10-10
Go底層channel實現原理及示例詳解
這篇文章主要介紹了Go底層channel實現原理及示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪
2022-08-08
用go寫的五子棋預測算法的實現
這篇文章主要介紹了用go寫的五子棋預測算法的實現，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2019-12-12
go mod包拉不下來的問題及解決方案
這篇文章主要介紹了go mod包拉不下來的問題及解決方案,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教
2023-09-09
Go語言bufio庫的全面指南與實戰(zhàn)技巧詳解
這篇文章主要為大家全面介紹一下?bufio?庫的核心組件與功能,包括?Reader、Writer?和?Scanner?等并深入探討它們在實際編程中的運用場景和技巧,感興趣的可以了解下
2024-01-01
Go結構體從基礎到應用深度探索
本文深入探討了結構體的定義、類型、字面量表示和使用方法,旨在為讀者呈現Go結構體的全面視角,通過結構體,開發(fā)者可以實現更加模塊化、高效的代碼設計,這篇文章旨在為您提供關于結構體的深入理解,助您更好地利用Go語言的強大功能
2023-10-10
Golang回調函數與閉包和接口函數的定義及使用介紹
這篇文章主要介紹了Golang回調函數與閉包和接口函數的定義及使用，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習吧
2023-05-05
教你一分鐘配置好Go語言開發(fā)環(huán)境(多種操作系統)
在這篇文章中,我們從頭到尾一步步指導你配置Golang開發(fā)環(huán)境,并編寫你的第一個"Hello,?World!"程序,我們詳細解釋了在多種操作系統（包括Windows、Linux和macOS）下的安裝過程、環(huán)境變量設置以及如何驗證安裝是否成功
2023-09-09
Golang defer延遲語句的實現
defer擁有注冊延遲調用的機制,本文主要介紹了Golang defer延遲語句的實現,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
2024-07-07