columnar

Documentation

Overview

Package columnar converts dataset.Dataset values to Apache Arrow and Parquet. Flattened ULog field paths become nullable column names, character arrays become UTF-8 strings, and the ULog format name, definition, and multi ID are preserved as Arrow schema metadata.

Index

• func ToArrow(source *dataset.Dataset, allocator memory.Allocator) (arrowlib.RecordBatch, error)
• func WriteParquet(destination io.Writer, source *dataset.Dataset) error

Examples

• ToArrow
• WriteParquet

Functions

func ToArrow

func ToArrow(source *dataset.Dataset, allocator memory.Allocator) (arrowlib.RecordBatch, error)

ToArrow converts source into one Arrow record batch. The caller owns the returned batch and must call its Release method. A nil allocator uses memory.DefaultAllocator.

Example

package main

import (
	"bytes"
	"fmt"

	"github.com/sunfish-robotics/ulog"
	"github.com/sunfish-robotics/ulog/pkg/columnar"
	"github.com/sunfish-robotics/ulog/pkg/dataset"
)

type sample struct {
	Timestamp uint64
	Pressure  float32
}

func main() {
	dataset := exampleDataset()
	record, err := columnar.ToArrow(dataset, nil)
	if err != nil {
		panic(err)
	}
	defer record.Release()

	fmt.Printf("%d rows, %d columns\n", record.NumRows(), record.NumCols())

}

func exampleDataset() *dataset.Dataset {
	var source bytes.Buffer
	writer, err := ulog.NewWriter(&source)
	if err != nil {
		panic(err)
	}
	stream, err := ulog.Register[sample](writer)
	if err != nil {
		panic(err)
	}
	for _, value := range []sample{
		{Timestamp: 1_000, Pressure: 101.25},
		{Timestamp: 2_000, Pressure: 101.5},
	} {
		if err := stream.Write(value); err != nil {
			panic(err)
		}
	}
	if err := writer.Close(); err != nil {
		panic(err)
	}
	file, err := dataset.Read(bytes.NewReader(source.Bytes()))
	if err != nil {
		panic(err)
	}
	dataset, err := file.Dataset("sample", 0)
	if err != nil {
		panic(err)
	}
	return dataset
}

Output:
2 rows, 2 columns

func WriteParquet

func WriteParquet(destination io.Writer, source *dataset.Dataset) error

WriteParquet writes source as one Parquet table using the same schema mapping as ToArrow. It does not close destination.

Example

package main

import (
	"bytes"
	"fmt"

	"github.com/sunfish-robotics/ulog"
	"github.com/sunfish-robotics/ulog/pkg/columnar"
	"github.com/sunfish-robotics/ulog/pkg/dataset"
)

type sample struct {
	Timestamp uint64
	Pressure  float32
}

func main() {
	dataset := exampleDataset()
	var destination bytes.Buffer
	if err := columnar.WriteParquet(&destination, dataset); err != nil {
		panic(err)
	}

	fmt.Printf("%s ... %s\n", destination.Bytes()[:4], destination.Bytes()[destination.Len()-4:])

}

func exampleDataset() *dataset.Dataset {
	var source bytes.Buffer
	writer, err := ulog.NewWriter(&source)
	if err != nil {
		panic(err)
	}
	stream, err := ulog.Register[sample](writer)
	if err != nil {
		panic(err)
	}
	for _, value := range []sample{
		{Timestamp: 1_000, Pressure: 101.25},
		{Timestamp: 2_000, Pressure: 101.5},
	} {
		if err := stream.Write(value); err != nil {
			panic(err)
		}
	}
	if err := writer.Close(); err != nil {
		panic(err)
	}
	file, err := dataset.Read(bytes.NewReader(source.Bytes()))
	if err != nil {
		panic(err)
	}
	dataset, err := file.Dataset("sample", 0)
	if err != nil {
		panic(err)
	}
	return dataset
}

Output:
PAR1 ... PAR1