Golang處理parquet文件實戰(zhàn)指南
前言
Parquet是Apache基金會支持的項目，是面向列存儲二進(jìn)制文件格式。支持不同類型的壓縮方式，廣泛用于數(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,用于說明struct中每個字段在生成parquet過程中如何被處理。
parquet-go包可以處理parquet數(shù)據(jù)，更多的tag可以參考其官網(wǎng)。
生成parquet文件
下面現(xiàn)給出生成parquet文件的代碼，然后分別進(jìn)行說明：
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)說明，但需要提醒的是類型與文檔保持一致：
| 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,用于寫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è)置壓縮類型，并通過defer操作確保關(guān)閉文件。下面開始寫數(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)寫數(shù)據(jù)，最后調(diào)用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
}
大概流程如下：首先定義本地文件，然后構(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)存可能有問題。這是官方文檔提示：
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。實際應(yīng)用中應(yīng)該分頁讀取，下面通過代碼進(jìn)行說明：
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ù)包括兩個參數(shù)，用于指定頁大小和頁數(shù)，關(guān)鍵代碼是跳過一定記錄：
pr.SkipRows(int64(pageSize * page))
根據(jù)這個方法可以獲得總行數(shù)，pr.GetNumRows()，然后結(jié)合頁大小計算總頁數(shù)，最后循環(huán)可以實現(xià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
}
首先打開文件，然后調(diào)用pr.GetNumRows()方法獲取總行數(shù)。然后基于路徑指定列，其中parquet_go_root為根路徑，因為前面使用字節(jié)數(shù)組，這里分割符變?yōu)閈u0001，完整路徑為：parquet_go_root\u0001score。
總結(jié)
到此這篇關(guān)于Golang處理parquet文件的文章就介紹到這了,更多相關(guān)Golang處理parquet文件內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Golang標(biāo)準(zhǔn)庫之errors包應(yīng)用方式
Go語言的errors包提供了基礎(chǔ)的錯誤處理能力,允許通過errors.New創(chuàng)建自定義error對象,error在Go中是一個接口,通過實現(xiàn)Error方法來定義錯誤文本,對錯誤的比較通?；趯ο蟮刂?而非文本內(nèi)容,因此即使兩個錯誤文本相同2024-10-10
這篇文章主要為大家詳細(xì)介紹了Golang中字符串的使用，文中的示例代碼講解詳細(xì)，對我們學(xué)習(xí)Golang有一定的幫助，感興趣的小伙伴可以了解一下2022-10-10
反射和元編程是一些高級編程概念，它們使開發(fā)者能夠在運行時檢查、修改并控制程序的行為，了解反射和元編程的工作方式可以幫助我們更好地理解Go，以及如何在需要的時候高效地使用它們，文章中介紹的非常詳細(xì)，感興趣的同學(xué)可以參考下2023-05-05
這篇文章主要介紹了如何使用Go來實現(xiàn)優(yōu)雅重啟服務(wù)，本文給大家介紹的非常詳細(xì)，具有一定的參考借鑒價值,需要的朋友可以參考下2019-11-11
這篇文章主要介紹了Go語言實現(xiàn)牛頓法求平方根函數(shù)的案例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-12-12
Go語言開發(fā)環(huán)境搭建與初探(Windows平臺下)
Go是Google開發(fā)的一種編譯型，並發(fā)型，并具有垃圾回收功能的編程語言,可能很多人想學(xué)習(xí)go語言，那么首先就要了解go語言的環(huán)境配置方法2014-10-10
這篇文章主要給大家介紹了Go語言中如何使用gorm，文中介紹的很詳細(xì)，有需要的朋友們可以參考借鑒，下面來一起看看吧。2016-12-12
這篇文章主要為大家詳細(xì)介紹了Go語言中定時任務(wù)cron的設(shè)計與使用,文中的示例代碼講解詳細(xì),對我們深入掌握Go語言有一定的幫助,需要的可以參考下2023-11-11
這篇文章主要介紹了golang 并發(fā)求和的實現(xiàn)方式，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-05-05
go語言開發(fā)環(huán)境安裝及第一個go程序(推薦)
這篇文章主要介紹了go語言開發(fā)環(huán)境安裝及第一個go程序，這篇通過實例代碼給大家介紹的非常詳細(xì)，具有一定的參考借鑒價值,需要的朋友可以參考下2020-02-02