org.wikimedia.eventutilities.flink.table

Class SchemaBuilderUtil

java.lang.Object

org.wikimedia.eventutilities.flink.table.SchemaBuilderUtil

public final class SchemaBuilderUtil
extends Object

Helper functions for creating and adding to Flink Schema.Builder for WMF event streams.

Method Summary

Modifier and Type	Method and Description
static org.apache.flink.table.api.Schema.Builder	createEventStreamSchemaBuilder(EventStream eventStream) Given an EventStream, converts its latest schema to Flink columns and returns a new Schema.Builder with all columns.
static org.apache.flink.table.api.Schema.Builder	createEventStreamSchemaBuilder(EventStream eventStream, String schemaVersion) Given an EventStream, converts its schema at the given version to Flink columns and returns a new Schema.Builder with all columns.
static org.apache.flink.table.api.Schema.Builder	withKafkaTimestampAsWatermark(org.apache.flink.table.api.Schema.Builder schemaBuilder) Adds a "kafka_timestamp" column to the schema builder and uses it as the watermark field with a default watermark delay of 10 seconds.
static org.apache.flink.table.api.Schema.Builder	withKafkaTimestampAsWatermark(org.apache.flink.table.api.Schema.Builder schemaBuilder, String columnName, int watermarkDelaySeconds) Adds kafka timestamp as a virtual column to the schema builder and uses it as the watermark with a delay of watermarkDelaySeconds.
static org.apache.flink.table.api.Schema.Builder	withKafkaTimestampAsWatermark(org.apache.flink.table.api.Schema.Builder schemaBuilder, String columnName, String sqlExpression) Adds kafka timestamp as a virtual column to the schema builder and uses it as the watermark.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Detail

createEventStreamSchemaBuilder

public static org.apache.flink.table.api.Schema.Builder createEventStreamSchemaBuilder(EventStream eventStream)

Given an EventStream, converts its latest schema to Flink columns and returns a new Schema.Builder with all columns.

createEventStreamSchemaBuilder

public static org.apache.flink.table.api.Schema.Builder createEventStreamSchemaBuilder(EventStream eventStream,
                                                                                       String schemaVersion)

Given an EventStream, converts its schema at the given version to Flink columns and returns a new Schema.Builder with all columns.

withKafkaTimestampAsWatermark

@Nonnull
public static org.apache.flink.table.api.Schema.Builder withKafkaTimestampAsWatermark(org.apache.flink.table.api.Schema.Builder schemaBuilder)

Adds a "kafka_timestamp" column to the schema builder and uses it as the watermark field with a default watermark delay of 10 seconds.

withKafkaTimestampAsWatermark

@Nonnull
public static org.apache.flink.table.api.Schema.Builder withKafkaTimestampAsWatermark(org.apache.flink.table.api.Schema.Builder schemaBuilder,
                                                                                               String columnName,
                                                                                               int watermarkDelaySeconds)

Adds kafka timestamp as a virtual column to the schema builder and uses it as the watermark with a delay of watermarkDelaySeconds.

Parameters:

columnName - Name of the timestamp column to add to the schema.

watermarkDelaySeconds - Seconds to delay the watermark by.

withKafkaTimestampAsWatermark

@Nonnull
public static org.apache.flink.table.api.Schema.Builder withKafkaTimestampAsWatermark(org.apache.flink.table.api.Schema.Builder schemaBuilder,
                                                                                               String columnName,
                                                                                               String sqlExpression)

Adds kafka timestamp as a virtual column to the schema builder and uses it as the watermark. See also Schema.Builder#watermark(String, String)

Parameters:

columnName - Name of the column to add to the schema.

sqlExpression - SQL expression to use for the watermark. This should probably refer to the columnName, e.g. "$columnName - INTERVAL '30' seconds"