streamio

package module
v0.0.8 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 11, 2026 License: MIT Imports: 15 Imported by: 0

README

streamio

CI Go Reference

A Go library and CLI for converting between NDJSON, Parquet, CSV, TSV, and Arrow IPC at scale, built for pipelines that move large datasets between storage and processing systems rather than one-off scripting. Any format can be read and written, decoding and dispatch run concurrently on a shared worker pool, and when the input and output formats match, bytes are streamed straight through with no decode step at all — reliably processing files far larger than available memory, fast.

Use cases

  • Bulk-loading columnar files into a datastore.
  • Converting exported NDJSON logs into columnar formats for analytics.
  • Reshaping CSV/TSV exports into a format a downstream system expects.
  • Any pipeline where the bottleneck is decode/encode throughput on multi-gigabyte files, not one-time convenience.

Install

Prebuilt CLI binaries for Linux and macOS (amd64/arm64) and Windows (amd64) are on the releases page. Download the archive for your platform, extract it, and put the streamio binary on your PATH.

To build from source instead:

go install github.com/AlexAslan/streamio/cmd/streamio@latest

Getting started

Convert a file (output format defaults to json; input format and compression are detected from the file's contents, with the extension only breaking ties):

./streamio convert --in data.ndjson --out data.parquet --out-format parquet

Compressed inputs (gzip, zstd, bzip2, xz, LZ4, Snappy, Brotli) are decompressed transparently, and an --out ending in .gz, .zst, .xz, .lz4, .sz or .br is compressed the same way. bzip2 is input-only, and Brotli (which has no magic number) is recognized by its .br extension:

./streamio convert --in data.csv.gz --out data.json.zst --csv-has-header

Profile a file in one pass: record count, how often each attribute is non-empty, and its most frequent values:

./streamio stats --in data.parquet --top 5
./streamio stats --in data.csv --csv-has-header --columns country,plan --output json

--out-format accepts json, parquet, csv, tsv, or arrow. See ./streamio convert --help for every flag, including CSV/TSV delimiter and header options, batch/worker sizing, and --max-row-errors.

As a library:

go get github.com/AlexAslan/streamio

import "github.com/AlexAslan/streamio"

result, err := streamio.ProcessFile(ctx, "data.parquet", handler,
    streamio.WithOutputFormat(streamio.FormatJSON),
)

See Configuration for the full streamio.Option reference.

Docs

  • Architecture — entry point, routing, and the decode/dispatch pool.
  • Formats — NDJSON, Parquet, and Arrow IPC read/write routes.
  • record/formatio — the generic cross-format seam every format package implements.
  • Configuration — streamio.Option reference and non-file-path input.
  • Benchmarks — measured throughput and memory across every format pair.
  • Package layout — where everything lives.

Contributing

See CONTRIBUTING.md. Security issues: SECURITY.md.

License

MIT. Release archives also include THIRD_PARTY_NOTICES.md with the licenses of bundled dependencies.

Documentation

Overview

Package streamio converts between document stream formats (NDJSON, Parquet), decoding an input file into canonical records and re-encoding it in the requested output format.

Index

Constants

View Source 
const (
	// FormatJSON selects newline-delimited JSON.
	FormatJSON = options.FormatJSON
	// FormatParquet selects Apache Parquet.
	FormatParquet = options.FormatParquet
	// FormatCSV selects comma-separated values.
	FormatCSV = options.FormatCSV
	// FormatTSV selects tab-separated values.
	FormatTSV = options.FormatTSV
	// FormatArrow selects Apache Arrow IPC (the file/random-access variant).
	FormatArrow = options.FormatArrow
)
View Source 
const (
	// CompressionAuto detects the compression from the file's leading bytes.
	CompressionAuto = options.CompressionAuto
	// CompressionNone reads the file as-is.
	CompressionNone = options.CompressionNone
	// CompressionGzip reads the file as a gzip stream.
	CompressionGzip = options.CompressionGzip
	// CompressionZstd reads the file as a Zstandard stream.
	CompressionZstd = options.CompressionZstd
	// CompressionBzip2 reads the file as a bzip2 stream.
	CompressionBzip2 = options.CompressionBzip2
	// CompressionXZ reads the file as an xz stream.
	CompressionXZ = options.CompressionXZ
	// CompressionLZ4 reads the file as an LZ4 frame stream.
	CompressionLZ4 = options.CompressionLZ4
	// CompressionSnappy reads the file as a framed Snappy stream.
	CompressionSnappy = options.CompressionSnappy
	// CompressionBrotli reads the file as a Brotli stream; never auto-detected from contents.
	CompressionBrotli = options.CompressionBrotli
)

Variables

View Source 
var ErrDecompressedTooLarge = decompress.ErrTooLarge

ErrDecompressedTooLarge is returned when a compressed input expands past WithMaxDecompressedSize.

View Source 
var ErrNoConversionPath = errors.New("streamio: no conversion path")

ErrNoConversionPath is returned when the requested output format can't be produced from the input file's native format: the pair needs a decoder for the input and an encoder for the output, and at least one of the two isn't registered. Wrapped with both formats by ProcessFile.

View Source 
var ErrTooManyRowErrors = options.ErrTooManyRowErrors

ErrTooManyRowErrors is TooManyRowErrorsError's sentinel: check for this failure with errors.Is without depending on the concrete type.

View Source 
var ErrUnknownFormat = sniff.ErrUnknownFormat

ErrUnknownFormat is returned when an input's contents match no supported format and its extension offers no hint either.

Functions

This section is empty.

Types

type ColumnStats added in v0.0.8 

type ColumnStats struct {
	// Name is the attribute path; nested map entries are dotted.
	Name string `json:"name"`
	// TopValues are the most frequent non-empty values, most frequent first.
	TopValues []ValueCount `json:"top_values"`
	// NonEmpty counts the records where the attribute is present and neither null nor empty.
	NonEmpty int64 `json:"non_empty"`
	// Pct is NonEmpty as a percentage (0–100) of all records.
	Pct float64 `json:"pct"`
	// Distinct is how many distinct non-empty values were counted.
	Distinct int64 `json:"distinct"`
	// DistinctExact is false once the attribute exceeded WithProfileMaxDistinct, making Distinct a
	// lower bound and the TopValues counts estimates (see ValueCount.MaxError).
	DistinctExact bool `json:"distinct_exact"`
}

ColumnStats summarizes one attribute of a profiled file.

type Compression added in v0.0.8 

type Compression = options.Compression

Compression names the stream compression wrapped around an input file.

type DocumentHandler 

type DocumentHandler func(context.Context, []byte) error

DocumentHandler is invoked once per output document. It may be called concurrently by multiple workers, so implementations must be safe for that.

type Format 

type Format = options.OutputFormat

Format identifies an input or output document encoding.

func DetectFormat added in v0.0.8 

func DetectFormat(src Source) (Format, error)

DetectFormat returns src's format from its contents, seeing through compression. The name's extension is only consulted when the contents leave the answer open.

func DetectInputFormat 

func DetectInputFormat(path string) Format

DetectInputFormat returns the format a file's name suggests, from its extension alone (a trailing .gz or .zst is ignored): any extension not named by a supported format means NDJSON.

ProcessFile itself sniffs the contents instead; prefer DetectFormat when the file is at hand.

type Logger 

type Logger = options.Logger

Logger receives ProcessFile diagnostics.

type Option 

type Option = options.Option

Option configures an optional aspect of a ProcessFile run.

func WithBatchSize 

func WithBatchSize(n int) Option

WithBatchSize sets how many documents are batched together per dispatch. n<=0 falls back to the default.

func WithCSVDelimiter 

func WithCSVDelimiter(r rune) Option

WithCSVDelimiter sets the field delimiter CSV or TSV reads and writes. Pass ',' for CSV or '\t' for TSV; the zero value falls back to the format's own default.

func WithCSVHasHeader 

func WithCSVHasHeader(b bool) Option

WithCSVHasHeader says whether a CSV or TSV file's first row names its columns rather than holding data. It defaults to false.

func WithChunkSize 

func WithChunkSize(n int) Option

WithChunkSize sets the fixed work-queue chunk size, in bytes, used to split NDJSON input across workers. n<=0 falls back to the default.

func WithInputCompression added in v0.0.8 

func WithInputCompression(c Compression) Option

WithInputCompression sets the source file's compression explicitly, instead of letting ProcessFile detect gzip or zstd from the file's leading bytes.

func WithInputFormat 

func WithInputFormat(f Format) Option

WithInputFormat sets the source file's format explicitly, instead of letting ProcessFile detect it from the file's contents.

func WithLogger 

func WithLogger(l Logger) Option

WithLogger sets the logger that receives ProcessFile diagnostics.

func WithMaxDecompressedSize added in v0.0.8 

func WithMaxDecompressedSize(n int64) Option

WithMaxDecompressedSize caps how many bytes a compressed input may expand to, guarding against decompression bombs. n<=0 means no cap.

func WithMaxOpenReaders 

func WithMaxOpenReaders(n int) Option

WithMaxOpenReaders caps the number of simultaneously open Parquet row-group readers. n<=0 falls back to the configured number of parallel workers, not a fixed default.

func WithMaxRowErrors 

func WithMaxRowErrors(n int, onSkip func(err error)) Option

WithMaxRowErrors caps how many rows may fail to decode and be skipped before the run fails, mirroring BigQuery's load-job max_bad_records: n<=0 means fail immediately on the first row error (the default), matching streamio's original behavior. onSkip, if non-nil, is called once per row skipped under the cap. Only NDJSON and CSV/TSV field errors count against the cap — a CSV/TSV syntax error and every Parquet decode error always fail the run regardless of n.

Exceeding the cap fails the run with a *TooManyRowErrorsError wrapping every row error collected up to and including the one that exceeded it; Result.Stats.RowsSkipped is still populated on that failure path.

func WithOutputFormat 

func WithOutputFormat(f Format) Option

WithOutputFormat sets the format documents are delivered in.

func WithParallelWorkers 

func WithParallelWorkers(n int) Option

WithParallelWorkers sets the number of concurrent decode workers. n<=0 falls back to the default.

func WithProfileColumns added in v0.0.8 

func WithProfileColumns(names ...string) Option

WithProfileColumns restricts ProfileFile to the named attributes and everything nested under them, addressing nested map entries with dots like WithTransforms.

func WithProfileMaxDistinct added in v0.0.8 

func WithProfileMaxDistinct(n int) Option

WithProfileMaxDistinct caps how many distinct values each decode worker counts exactly per attribute, bounding memory on high-cardinality columns. n<=0 falls back to the default of 100,000.

func WithProfileTopN added in v0.0.8 

func WithProfileTopN(n int) Option

WithProfileTopN sets how many most-frequent values ProfileFile keeps per attribute. n<=0 falls back to the default of 10.

func WithReadBufferSize 

func WithReadBufferSize(n int) Option

WithReadBufferSize sets the read buffer size, in bytes, used while scanning the input file. n<=0 falls back to the default.

func WithSingleFileOutput 

func WithSingleFileOutput(b bool) Option

WithSingleFileOutput requests one continuous output document spanning the whole run, for formats that support it, instead of one document per batch. Forces decoding to a single worker.

func WithTransforms 

func WithTransforms(rules ...TransformRule) Option

WithTransforms applies a set of rename/drop rules to every decoded record before it is encoded into the requested output format.

type Profile added in v0.0.8 

type Profile struct {
	// Columns lists every attribute in first-seen order.
	Columns []ColumnStats `json:"columns"`
	// Records is the number of records read.
	Records int64 `json:"records"`
	// RowsSkipped is how many rows WithMaxRowErrors allowed to be skipped; they are not in Records.
	RowsSkipped int64 `json:"rows_skipped"`
	// ColumnsTruncated is true when the file had more distinct attribute paths than are tracked.
	ColumnsTruncated bool `json:"columns_truncated"`
}

Profile is the result of ProfileFile.

func ProfileFile added in v0.0.8 

func ProfileFile(ctx context.Context, path string, opts ...Option) (Profile, error)

ProfileFile reads the file at path and summarizes it: the record count, how often each attribute is non-empty, and its most frequent values.

An attribute is empty when it is missing, null, a zero-length string, or an empty list; whitespace is not empty. Nested entries are named by dotted path, and an attribute repeated within one record counts once, from its first occurrence. A value counts under the text it renders as, so the number 1 and the string "1" are one value, and a list counts under the single value "[list]".

Each decode worker tracks up to WithProfileMaxDistinct values per attribute exactly, then falls back to approximate heavy-hitter counting; non-empty counts stay exact either way. Columns are reported in first-seen order, which for NDJSON whose keys differ between records can vary with the worker count.

func ProfileReaderAt added in v0.0.8 

func ProfileReaderAt(ctx context.Context, src Source, opts ...Option) (Profile, error)

ProfileReaderAt summarizes src the same way ProfileFile does for an on-disk file.

Every input goes through the same canonical-record path ProcessFile uses, so equal data gives an equal profile whichever format or compression it arrives in. See ProfileFile for what counts as empty and how attributes are named.

type Result 

type Result = options.Result

Result reports the outcome of a ProcessFile run.

func ProcessFile 

func ProcessFile(
	ctx context.Context,
	path string,
	handler DocumentHandler,
	opts ...Option,
) (Result, error)

ProcessFile reads the file at path and calls handler once per document, delivering each one in cfg.OutputFormat (options.FormatJSON unless WithOutputFormat says otherwise); see Option for tuning.

func ProcessReaderAt 

func ProcessReaderAt(
	ctx context.Context,
	src Source,
	handler DocumentHandler,
	opts ...Option,
) (Result, error)

ProcessReaderAt reads src and calls handler once per document, the same way ProcessFile does for an on-disk file — src need not be backed by a real file, only support ReadAt over its declared Size, which is what every format's chunked-parallel decoder actually requires.

type Source 

type Source = options.Source

Source is a sized, randomly-addressable byte source ProcessReaderAt reads from: an *os.File, an in-memory buffer, or anything else providing ReadAt over a known span. Name labels it in diagnostics and error text and need not be a real file path.

type Stats 

type Stats = options.Stats

Stats holds counters describing a completed ProcessFile run.

type TooManyRowErrorsError 

type TooManyRowErrorsError = options.TooManyRowErrorsError

TooManyRowErrorsError is returned when WithMaxRowErrors' limit is exceeded: it wraps every row error collected up to and including the one that exceeded it.

type TransformRule 

type TransformRule = options.PathTransformRule

TransformRule is a single rename or drop rule built by RenamePath or DropPath.

func DropPath 

func DropPath(path string) TransformRule

DropPath builds a transform rule that removes one field path.

func RenamePath 

func RenamePath(from, to string) TransformRule

RenamePath builds a transform rule that renames one field path.

type ValueCount added in v0.0.8 

type ValueCount struct {
	// Value is the value rendered as text, the way the converters render it.
	Value string `json:"value"`
	// Count is how many records held the value; an estimate when MaxError is non-zero.
	Count int64 `json:"count"`
	// MaxError bounds how far Count may be from the true count. It is zero unless the attribute
	// exceeded WithProfileMaxDistinct.
	MaxError int64 `json:"max_error,omitempty"`
	// Pct is Count as a percentage (0–100) of all records.
	Pct float64 `json:"pct"`
}

ValueCount is one value and how many records held it.

Source Files

View all Source files

Directories

Path Synopsis
cmd
streamio command
Command streamio-cli converts a document file between the formats streamio supports (NDJSON, Parquet, CSV, TSV, and Arrow IPC), and profiles them.
 Command streamio-cli converts a document file between the formats streamio supports (NDJSON, Parquet, CSV, TSV, and Arrow IPC), and profiles them.
internal
arrowio
Package arrowio supplies streamio's Apache Arrow IPC (file format) capabilities: decoding into canonical records and encoding records back into Arrow IPC documents.
 Package arrowio supplies streamio's Apache Arrow IPC (file format) capabilities: decoding into canonical records and encoding records back into Arrow IPC documents.
csvio
Package csvio owns streamio's CSV and TSV support: one package for both, since they differ only in field delimiter.
 Package csvio owns streamio's CSV and TSV support: one package for both, since they differ only in field delimiter.
decompress
Package decompress turns a compressed source (gzip, zstd, bzip2, xz, LZ4, Snappy, or Brotli) into the sized, randomly-addressable options.Source every format's decoder needs.
 Package decompress turns a compressed source (gzip, zstd, bzip2, xz, LZ4, Snappy, or Brotli) into the sized, randomly-addressable options.Source every format's decoder needs.
formatio
Package formatio defines the per-format capabilities streamio composes: streaming a file's own bytes (RawSource), decoding into canonical records (RecordDecoder), and encoding records as another format's documents (RecordEncoder).
 Package formatio defines the per-format capabilities streamio composes: streaming a file's own bytes (RawSource), decoding into canonical records (RecordDecoder), and encoding records as another format's documents (RecordEncoder).
jsonio
Package jsonio owns streamio's JSON output path: encoding canonical record.Records as JSON documents, and decoding NDJSON back into them.
 Package jsonio owns streamio's JSON output path: encoding canonical record.Records as JSON documents, and decoding NDJSON back into them.
options
Package options defines streamio's Config and the functional Option funcs that build it, along with the transform rules and format constants ProcessFile's callers configure a run with.
 Package options defines streamio's Config and the functional Option funcs that build it, along with the transform rules and format constants ProcessFile's callers configure a run with.
parquetio
Package parquetio supplies streamio's Parquet capabilities: raw row-group passthrough, decoding into canonical records, and encoding records back into Parquet documents.
 Package parquetio supplies streamio's Parquet capabilities: raw row-group passthrough, decoding into canonical records, and encoding records back into Parquet documents.
pool
Package pool runs the decode/dispatch worker pool shared by every streamio route: each format only supplies a formatio.RawSource (RunRaw) or a decoder/encoder pair (RunRecords).
 Package pool runs the decode/dispatch worker pool shared by every streamio route: each format only supplies a formatio.RawSource (RunRaw) or a decoder/encoder pair (RunRecords).
profile
Package profile summarizes a stream of canonical records: how many there are, how often each attribute is non-empty, and which values occur most.
 Package profile summarizes a stream of canonical records: how many there are, how often each attribute is non-empty, and which values occur most.
record
Package record defines streamio's canonical, format-neutral row: the intermediate every cross-format conversion goes through between a decoder and an encoder.
 Package record defines streamio's canonical, format-neutral row: the intermediate every cross-format conversion goes through between a decoder and an encoder.
sniff
Package sniff identifies an input's compression and document format from its leading bytes, using the file extension only as a hint for inputs the content can't settle.
 Package sniff identifies an input's compression and document format from its leading bytes, using the file extension only as a hint for inputs the content can't settle.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.