org.wikimedia.eventutilities.flink.table

Class EventTableDescriptorBuilder

java.lang.Object

org.wikimedia.eventutilities.flink.table.EventTableDescriptorBuilder

@ParametersAreNonnullByDefault
public class EventTableDescriptorBuilder
extends Object

Builder wrapper to aid in constructing Flink TableDescriptors using Wikimedia Event Streams. TableDescriptors are passed to the Flink TableEnvironment when instantiating Table objects or when creating tables in a catalog. This class uses EventStreamFactory to get information about Event Streams declared in EventStreamConfig, e.g. schema and topics. You must at minimum call eventStream() and connector() before calling build(). It is possible to pass in your own Flink Schema.Builder while using this wrapper. If you need to augment the schema, for example to add computed / virtual fields, calling getSchemaBuilder() will get you the Schema.Builder with the stream's schema set as the physical DataType. After modifying the Schema.Builder as desired, call schemaBuilder(modifiedSchemaBuilder) again to use it for the TableDescriptor that will be built. Examples: Instantiate a EventTableDescriptorBuilder from URIs:

     EventTableDescriptorBuilder builder = EventTableDescriptorBuilder.from(
          Arrays.asList(
              "https://schema.wikimedia.org/repositories/primary/jsonschema",
              "https://schema.wikimedia.org/repositories/secondary/jsonschema",
          ),
          "https://meta.wikimedia.org/w/api.php"
     );
 

Build a DataGen Table:

     Table t = tableEnv.from(
          builder
              .connector("datagen")
              .eventStream("mediawiki.page-create")
              .option("number-of-rows", "10")
              .build()
           )
     );
 

Build a Kafka Table:

     Table t = tableEnv.from(
          builder
              .eventStream("mediawiki.page-create")
              .setupKafka(
                  "localhost:9092",
                  "my_consumer_group",
                  true  // If should auto-add a kafka_timestamp watermark virtual field.
               )
              .build()
           )
     );
 

Constructor Summary

Constructors

Constructor and Description
EventTableDescriptorBuilder(EventStreamFactory eventStreamFactory)

Method Summary

Modifier and Type	Method and Description
org.apache.flink.table.api.TableDescriptor	build() Applies all the collected configs and options to build a TableDescriptor using information from the EventStream.
EventTableDescriptorBuilder	clear() Clears all collected configs and options from this EventTableDescriptorBuilder so that it can be used to build another TableDescriptor.
EventTableDescriptorBuilder	connector(String connectorIdentifier) Sets the connector.
EventTableDescriptorBuilder	eventStream(EventStream eventStream) Sets the EventStream that will be used to build the TableDescriptor.
EventTableDescriptorBuilder	eventStream(String streamName) Sets the EventStream by streamName that will be used to build the TableDescriptor.
EventTableDescriptorBuilder	format(String format) Sets the format for the TableDescriptor.Builder.
static EventTableDescriptorBuilder	from(List<String> eventSchemaBaseUris, String eventStreamConfigUri) EventTableDescriptorBuilder factory method.
static EventTableDescriptorBuilder	from(List<String> eventSchemaBaseUris, String eventStreamConfigUri, Map<String,String> httpClientRoutes) EventTableDescriptorBuilder factory method.
EventStreamFactory	getEventStreamFactory() Convenience method to get the EventStreamFactory used by this EventTableDescriptorBuilder.
org.apache.flink.table.api.Schema.Builder	getSchemaBuilder() Use this to get the schemaBuilder for the eventStream in order to augment and reset it before calling build().
EventTableDescriptorBuilder	option(String key, String value) Sets a single TableDescriptor.Builder option.
EventTableDescriptorBuilder	options(Map<String,String> options) Sets the TableDescriptor.Builder options.
EventTableDescriptorBuilder	partitionedBy(String... partitionKeys) Sets the partitionKeys for the TableDescriptor.Builder.
EventTableDescriptorBuilder	schemaBuilder(org.apache.flink.table.api.Schema.Builder schemaBuilder) Sets the Schema.Builder.
EventTableDescriptorBuilder	setupKafka(String bootstrapServers, String consumerGroup) Helper to aid in setting up a kafka connector for the EventStream.
EventTableDescriptorBuilder	withKafkaTimestampAsWatermark() Adds a "kafka_timestamp" column to the schema and uses it as the watermark field with a default watermark delay of 10 seconds.
EventTableDescriptorBuilder	withKafkaTimestampAsWatermark(String columnName, int watermarkDelaySeconds) Adds kafka timestamp as a virtual column to the schema, and uses it as the watermark with a delay of watermarkDelaySeconds.
EventTableDescriptorBuilder	withKafkaTimestampAsWatermark(String columnName, String sqlExpression) Adds kafka timestamp as a virtual column to the schema, and uses it as the watermark.

Methods inherited from class java.lang.Object

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

Constructor Detail

EventTableDescriptorBuilder

public EventTableDescriptorBuilder(EventStreamFactory eventStreamFactory)

Method Detail

from

@Nonnull
public static EventTableDescriptorBuilder from(List<String> eventSchemaBaseUris,
                                                        String eventStreamConfigUri)

EventTableDescriptorBuilder factory method.

Parameters:

eventSchemaBaseUris - URIs from which to fetch event JSONSchemas.

eventStreamConfigUri - URI from which to fetch event stream config.

from

@Nonnull
public static EventTableDescriptorBuilder from(List<String> eventSchemaBaseUris,
                                                        String eventStreamConfigUri,
                                                        Map<String,String> httpClientRoutes)

EventTableDescriptorBuilder factory method.

Parameters:

eventSchemaBaseUris - URIs from which to fetch event JSONSchemas.

eventStreamConfigUri - URI from which to fetch event stream config.

httpClientRoutes - Map of source to dest http client routes. E.g. "https://meta.wikimedia.org" to "https://api-ro.wikimedia.org" If null, no special routing will be configured.

getEventStreamFactory

@Nonnull
public EventStreamFactory getEventStreamFactory()

Convenience method to get the EventStreamFactory used by this EventTableDescriptorBuilder.

connector

@Nonnull
public EventTableDescriptorBuilder connector(String connectorIdentifier)

Sets the connector.

Parameters:

connectorIdentifier - A valid and registered Flink Table Connector Factory Identifier string. e.g. "kafka" or "filesystem".

eventStream

@Nonnull
public EventTableDescriptorBuilder eventStream(EventStream eventStream)

Sets the EventStream that will be used to build the TableDescriptor.

eventStream

@Nonnull
public EventTableDescriptorBuilder eventStream(String streamName)

Sets the EventStream by streamName that will be used to build the TableDescriptor.

Parameters:

streamName - Name of the EventStream, must be declared in EventStreamConfig.

schemaBuilder

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

Sets the Schema.Builder. If you don't set this the Table Schema will be converted from the eventStream's JSONSchema.

options

@Nonnull
public EventTableDescriptorBuilder options(Map<String,String> options)

Sets the TableDescriptor.Builder options. This clears out any previously set options.

option

@Nonnull
public EventTableDescriptorBuilder option(String key,
                                                   String value)

Sets a single TableDescriptor.Builder option.

format

@Nonnull
public EventTableDescriptorBuilder format(String format)

Sets the format for the TableDescriptor.Builder. This needed by most, but not all, connectors.

partitionedBy

@Nonnull
public EventTableDescriptorBuilder partitionedBy(String... partitionKeys)

Sets the partitionKeys for the TableDescriptor.Builder.

setupKafka

@Nonnull
public EventTableDescriptorBuilder setupKafka(String bootstrapServers,
                                                       String consumerGroup)

Helper to aid in setting up a kafka connector for the EventStream. You must first call eventStream() before calling this method so that it can fill in the details for the connector from the EventStream. Note: The Kafka Table Connector option scan.start.mode will not be set, relying on the default value of "group-offsets". The Kafka consumer property auto.offset.reset will be set to "latest", which will be used in the case that there are no committed group offsets. Override this behavior by calling e.g. option("properties.auto.offset.reset", "earliest") This does not set the key format or hoist any metadata fields (like kafka_timestamp) into the schema. See withKafkaTimestampAsWatermark to help use the kafka timestamp as the watermark.

Parameters:

bootstrapServers - Kafka bootstrap.servers property.

consumerGroup - Kafka consumer.group.id property.

withKafkaTimestampAsWatermark

@Nonnull
public EventTableDescriptorBuilder withKafkaTimestampAsWatermark()

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

withKafkaTimestampAsWatermark

@Nonnull
public EventTableDescriptorBuilder withKafkaTimestampAsWatermark(String columnName,
                                                                          int watermarkDelaySeconds)

Adds kafka timestamp as a virtual column to the schema, 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 EventTableDescriptorBuilder withKafkaTimestampAsWatermark(String columnName,
                                                                          String sqlExpression)

Adds kafka timestamp as a virtual column to the schema, 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"

getSchemaBuilder

@Nonnull
public org.apache.flink.table.api.Schema.Builder getSchemaBuilder()

Use this to get the schemaBuilder for the eventStream in order to augment and reset it before calling build(). If you change the Schema.Builder, make sure you call schemaBuilder(modifiedSchemaBuilder) to apply your changes before calling build().

Returns:

The Schema.Builder that will be used to build the Table Schema.

build

@Nonnull
public org.apache.flink.table.api.TableDescriptor build()

Applies all the collected configs and options to build a TableDescriptor using information from the EventStream. You can use the returned TableDescriptor to instantiate a Table object or create a Table in a catalog. Any collected state will be cleared before returning, so you can re-use this EventTableDescriptorBuilder again.

clear

@Nonnull
public final EventTableDescriptorBuilder clear()

Clears all collected configs and options from this EventTableDescriptorBuilder so that it can be used to build another TableDescriptor.