Parquet-encoder

The encoder consumes a stream of events and encodes them into parquet record. All records are stored in local files in configured directory. When file size or record count constraint is reached the encoder closes the local file and emits it to next operator for further processing.

JAVAYAML

.processWith(ParquetEncoderConfigBuilder.builder()
    .withDirectory(value)
    .withPrefix(value)
    .withMaxFileSize(builder -> builder
        .withBytes(value)
    )
    .withMaxRecords(value)
    .withCloseOnIdleTimeout(value)
    .withWriter(builder -> builder
        .withRecordClass(value)
        .withRowGroupSize(builder -> builder
            .withBytes(value)
        )
        .withPageSize(builder -> builder
            .withBytes(value)
        )
        .withCompression(value)
        .withCompressionLevel(value)
    )
)

processor:
  parquet-encoder:
    directory: value
    prefix: value
    maxFileSize:
      bytes: value
    maxRecords: value
    closeOnIdleTimeout: value
    writer:
      recordClass: value
      rowGroupSize:
        bytes: value
      pageSize:
        bytes: value
      compression: value
      compressionLevel: value

Java dependency management

Add this declaration to your dependency management system to access the configuration DSL for this plugin in Java.

MavenGradle

<dependency>
    <groupId>org.voltdb</groupId>
    <artifactId>volt-stream-plugin-parquet-api</artifactId>
    <version>1.8.7</version>
</dependency>

implementation group: 'org.voltdb', name: 'volt-stream-plugin-parquet-api', version: '1.8.7'

Properties

directory

Directory where parquet files will be stored Type: string

prefix

Prefix for parquet files Type: string

Default value: data

maxFileSize

Maximum file size in bytes before emitting Type: object

Default value: 128mb

Fields of maxFileSize:

maxFileSize.bytes

Type: number

maxRecords

Maximum number of records before rolling Type: number

Default value: 1000000

closeOnIdleTimeout

When stream is idle, encoder will close parquet file after configured timeout. 0 or negative value means to never close a file by a timer.

Type: object

Default value: PT0S

writer

Creates the parquet writer of each file. In YAML, the keys below configure the reflective writer; in Java, any ParquetWriterFactory implementation can be set.

Required.

Type: object

Fields of writer:

writer.recordClass

Fully qualified name of the class of the written records; use '$' for a nested class, for example com.example.Pipeline$Event. Every non-static field becomes a column and must be of type int, long, float, double, boolean, String or their boxed types.

Required.

Type: class

writer.rowGroupSize

Parquet row group size Type: object

Default value: 128mb

Fields of writer.rowGroupSize:

writer.rowGroupSize.bytes

Type: number

writer.pageSize

Parquet page size Type: object

Default value: 8kb

Fields of writer.pageSize:

writer.pageSize.bytes

Type: number

writer.compression

Compression codec: uncompressed, snappy or zstd Type: object

Supported values: uncompressed, snappy, zstd.

Default value: uncompressed

writer.compressionLevel

ZSTD compression level from 1 to 22. Valid only with zstd; when not set, ZSTD uses level 3. Type: number

JSON Schema

You can validate or explore the configuration using its JSON Schema.

Usage Examples

JAVAYAML

builder .consumeFromSource(...) .processWith(ParquetEncoderConfigBuilder.builder() .withDirectory("/parquet/output/path") .withMaxFileSize(MemoryUnit.ofBytes(96)) .withMaxRecords(100) .withCloseOnIdleTimeout(Duration.ofMinutes(1)) .withPrefix("events") .withWriter(ReflectiveWriterConfigBuilder.builder() .withRecordClass(Event.class) .withCompression(ParquetCompression.ZSTD) .build()) ) .processWith(path -> { ... })

processor: parquet-encoder: directory: /parquet/output/path maxRecords: 100 closeOnIdleTimeout: PT1M prefix: events writer: recordClass: com.example.Event compression: zstd compressionLevel: 19