Parquet логирование с помощью Golang - UPROGER

Все, кто работает с “большими данными”, знают о Apache Parquet как о решении для хранения данных. Parquet разработан для обеспечения производительности, масштабируемости и совместимости с различными системами обработки данных, что делает его отличным выбором для вашего ETL-конвейера.

Эта статья состоит из следующих разделов:

Раздел 1: Apache Parquet
Раздел 2: Сжатие паркетов
Раздел 3: Golang + Apache Parquet
Раздел 4: Заключение

Раздел 1: Apache Parquet

Official Documentation

Apache Parquet – это формат хранения данных в виде столбцов, который является отличной альтернативой традиционным форматам хранения данных на основе строк, таким как CSV, TSV и RDBMS. Он разработан для обеспечения производительности и масштабируемости, необходимых для конвейеров обработки Больших Данных.

Если вы не знаете, что такое “колоночное хранение”, рассмотрите пример сценария заказа хот-дога, в котором сравниваются форматы хранения на основе строк и на основе колонок:

На основе строк (CSV):

order_id,order_date,customer_name,product,order_price,order_quantity
1,2023-01-01,Chad,hotdog,1.00,1
2,2023-01-01,Jane,hotdog,1.00,3
3,2023-01-01,Jane,dietcoke,.50,3

Основанная на колоннах (паркет):

Примечание: Это упрощенный пример, реальный формат паркета намного сложнее.

<column order_id>
1
2
3

<column order_date>
2023-01-01
2023-01-01
2023-01-01

<column customer_name>
Chad
Jane
Jane

<column product>
hotdog
hotdog
dietcoke

<column order_price>
1.00
1.00
0.50

<column order_quantity>
1
3
3

Раздел 2: Сжатие данных parquet

Наиболее часто используемым алгоритмом сжатия данных parquet является SNAPPY, но parquet поддерживает множество алгоритмов сжатия.

SNAPPY: Библиотека быстрой компрессии и декомпрессии, разработанная в Google. Она разработана для скорости и особенно полезна в приложениях, где приоритетом является высокоскоростная передача данных.

GZIP: популярный и широко распространенный алгоритм сжатия, используемый для сжатия и распаковки файлов. Он медленнее, чем некоторые другие алгоритмы, но предлагает хороший баланс между скоростью сжатия и коэффициентом.

LZO: алгоритм сжатия на основе Lempel-Ziv (LZ), который использует кодирование по длине строки (RLE) и кодирование Хаффмана для достижения высокого коэффициента сжатия.

BROTLI: относительно новый алгоритм сжатия, разработанный компанией Google. Он разработан для достижения более высоких коэффициентов сжатия, чем gzip, при той же или более высокой скорости.

LZ4: очень быстрый алгоритм сжатия, который был разработан для скорости. Он особенно полезен в приложениях, где приоритетом является высокоскоростная передача данных, таких как коммуникации в реальном времени или приложения для работы с большими данными.

ZSTD: относительно новый алгоритм сжатия, разработанный для обеспечения хорошего баланса между скоростью и коэффициентом сжатия. Он особенно полезен в приложениях, где приоритетом является высокоскоростная передача данных.

Лично я предпочитаю использовать сжатие SNAPPY при работе с данными Parquet; однако стоит отметить, что оптимальный алгоритм сжатия может варьироваться в зависимости от конкретного случая использования. Для более глубокого понимания вы можете обратиться к официальной документации, которая обычно включает подробные эталонные показатели и сравнения каждого доступного алгоритма. Вот несколько надежных источников, которые могут оказаться полезными в ваших исследованиях:

Section 3: Golang + Apache Parquet

Существует 2 основные библиотеки, которые вы можете использовать для работы с файлами Parquet в Go:

xitongsys/parquet-go (1k stars on GitHub)
segmentio/parquet-go (319 start on GitHub)

Хотя обе библиотеки эффективно решают одну и ту же задачу, лично я предпочитаю использовать xitongsys/parquet-go. В целях последовательности я буду использовать эту библиотеку до конца этого обсуждения. Тем не менее, стоит подчеркнуть, что любая из библиотек представляет собой хороший выбор для работы с данными Parquet. Если ваши потребности более уникальны, чем то, что предлагают эти библиотеки, подумайте о том, чтобы сделать pull request или сделать форк своей собственной версии.

Давайте начнем с изучения того, как записывать в файл parquet. Стоит отметить, что в этом примере мы используем теги struct для определения схемы, а также сжатие Snappy. Важно помнить, что эти параметры полностью настраиваются при создании объекта parquet writer. Не стесняйтесь экспериментировать с различными конфигурациями, чтобы найти то, что лучше всего соответствует вашим потребностям.

// 1. Define the schema using go struct tags
type HotDogOrder struct {
	ID           int64   `parquet:"name=id, type=INT64"`
	Date         int32   `parquet:"name=date, type=INT32, convertedtype=TIME_MILLIS"`
	CustomerName string  `parquet:"name=customer_name, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
	Product      string  `parquet:"name=product, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
	Price        float64 `parquet:"name=price, type=DOUBLE"`
	Quantity     int     `parquet:"name=quantity, type=INT32"`
}

func main() {
  writeParquet()
}

func writeParquet() {
	var err error
	// 2. Create the output file
	parqFile, err := os.Create("hot_dog_orders.snappy.parquet")
	if err != nil {
		panic(err)
	}

	// 3. Create a Parquet file writer
	parqWriter, err := writer.NewParquetWriterFromWriter(parqFile, new(HotDogOrder), 4)
	if err != nil {
		panic(err)
	}

	// 4. Set the compression type (note: Snappy is already the default but this is how you can change it)
	parqWriter.CompressionType = parquet.CompressionCodec_SNAPPY

	// 5. Write some random data to the file
	for i := 0; i < 100; i++ {
		hdOrder := generateHotDogOrder(i)
		if err = parqWriter.Write(hdOrder); err != nil {
			panic(err)
		}
	}

	// 6. Close the parquet writer
	if err = parqWriter.WriteStop(); err != nil {
		panic(err)
	}

	// 7. Close the file
	if err = parqFile.Close(); err != nil {
		panic(err)
	}
}


func generateHotDogOrder(id int) HotDogOrder {
	products := []string{"Hot Dog", "Diet Coke"}
	customerNames := []string{"John Doe", "Jane Doe", "John Smith", "Jane Smith"}
	return HotDogOrder{
		ID:           int64(id),
		Date:         int32(time.Now().Unix() * 1000),
		CustomerName: customerNames[rand.Intn(len(customerNames))],
		Product:      products[rand.Intn(len(products))],
		Price:        0.50 + (rand.Float64() * (8.00 - 0.50)),
		Quantity:     rand.Intn(10),
	}
}

Чтение файлов parquet в Go относительно просто, однако важно отметить, что любые изменения, внесенные в данные parquet в памяти, не будут автоматически сохранены в исходном файле. Чтобы внести постоянные изменения, вам нужно будет записать новый файл parquet или перезаписать исходный.

// 1. Define the schema using go struct tags
type HotDogOrder struct {
	ID           int64   `parquet:"name=id, type=INT64"`
	Date         int32   `parquet:"name=date, type=INT32, convertedtype=TIME_MILLIS"`
	CustomerName string  `parquet:"name=customer_name, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
	Product      string  `parquet:"name=product, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
	Price        float64 `parquet:"name=price, type=DOUBLE"`
	Quantity     int     `parquet:"name=quantity, type=INT32"`
}

func readParquet() {
	// 2. Create the file reader pointed to the parquet file
	fileReader, err := local.NewLocalFileReader("hot_dog_orders.snappy.parquet")
	if err != nil {
		panic(err)
	}

	// 3. Create a parquet reader
	parqReader, err := reader.NewParquetReader(fileReader, new(HotDogOrder), 4)
	if err != nil {
		panic(err)
	}

	// 4. Create a slice to hold the data
	numRows := int(parqReader.GetNumRows())
	hdOrders := make([]HotDogOrder, numRows)

	// 5. Read the data from the file
	if err = parqReader.Read(&hdOrders); err != nil {
		panic(err)
	}

	// 6. Close the parquet reader and file reader
	parqReader.ReadStop()
	if err = fileReader.Close(); err != nil {
		panic(err)
	}

	// 7. Print the data
	for _, hdOrder := range hdOrders {
		fmt.Println(hdOrder.ID, hdOrder.Date, hdOrder.CustomerName, hdOrder.Product, hdOrder.Price, hdOrder.Quantity)
	}
}

func main() {
  readParquet()
}

Примечание: В производственной среде столкновение с проблемами не должно сразу приводить к панике (err). Скорее, рекомендуется реализовать систему повторных попыток или резервного копирования в качестве резервного механизма в случае возникновения проблем. Кроме того, лучшей практикой является отделение процесса протоколирования от основной рабочей нагрузки приложения с помощью goroutines и каналов.

Раздел 4: Заключение

Apache Parquet предоставляет высокоэффективный формат хранения данных, который может ускорить ваш конвейер ETL “Больших данных”. При работе с функциями writeParquet()и readParquet() из фрагментов кода доступно множество опций; например, вы можете читать и записывать данные непосредственно в облачное хранилище, конвертировать данные из CSV в Parquet, настраивать RowGroupSize или CompressionType файла parquet и многое другое.

Вам не обязательно полагаться на создание функции readParquet(), чтобы увидеть содержимое файла; ниже приведены некоторые из доступных вариантов просмотра и инспекции файлов parquet:

Используйте этот CLI-инструмент от автора parquet-go для быстрого просмотра файлов parquet из терминала.
Используйте Apache Hive для более традиционного представления базы данных. Вы можете использовать Hive, Impala и Spark для преобразования, трансформации и запросов к таблицам parquet.
Используйте плагин или расширение в вашей IDE. Я использую Avro и Parquet Viewer в GoLand или плагин Big Data Tools для DataGrip. Я не использую VSCode, но нашел это репозиторий на GitHub.

Спасибо за чтение!