org.apache.flink.streaming.connectors.kafka

Class KafkaTableSource

java.lang.Object

org.apache.flink.streaming.connectors.kafka.KafkaTableSource

All Implemented Interfaces:

org.apache.flink.table.sources.DefinedProctimeAttribute, org.apache.flink.table.sources.DefinedRowtimeAttributes, org.apache.flink.table.sources.StreamTableSource<Row>, org.apache.flink.table.sources.TableSource<Row>

Direct Known Subclasses:

Kafka010TableSource, Kafka011TableSource, Kafka08TableSource, Kafka09TableSource, KafkaAvroTableSource, KafkaJsonTableSource

@Internal
public abstract class KafkaTableSource
extends Object
implements org.apache.flink.table.sources.StreamTableSource<Row>, org.apache.flink.table.sources.DefinedProctimeAttribute, org.apache.flink.table.sources.DefinedRowtimeAttributes

A version-agnostic Kafka StreamTableSource.

The version-specific Kafka consumers need to extend this class and override createKafkaConsumer(String, Properties, DeserializationSchema)}.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	KafkaTableSource.Builder<T extends KafkaTableSource,B extends KafkaTableSource.Builder> Abstract builder for a KafkaTableSource to be extended by builders of subclasses of KafkaTableSource.

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	KafkaTableSource(String topic, Properties properties, org.apache.flink.table.api.TableSchema schema, TypeInformation<Row> returnType) Creates a generic Kafka StreamTableSource.

Method Summary

Modifier and Type	Method and Description
protected abstract FlinkKafkaConsumerBase<Row>	createKafkaConsumer(String topic, Properties properties, DeserializationSchema<Row> deserializationSchema) Creates a version-specific Kafka consumer.
boolean	equals(Object o)
String	explainSource()
DataStream<Row>	getDataStream(StreamExecutionEnvironment env) NOTE: This method is for internal use only for defining a TableSource.
protected abstract DeserializationSchema<Row>	getDeserializationSchema() Returns the deserialization schema.
protected FlinkKafkaConsumerBase<Row>	getKafkaConsumer(String topic, Properties properties, DeserializationSchema<Row> deserializationSchema) Returns a version-specific Kafka consumer with the start position configured.
String	getProctimeAttribute()
TypeInformation<Row>	getReturnType()
List<org.apache.flink.table.sources.RowtimeAttributeDescriptor>	getRowtimeAttributeDescriptors()
org.apache.flink.table.api.TableSchema	getTableSchema()
int	hashCode()
protected void	setProctimeAttribute(String proctimeAttribute) Declares a field of the schema to be the processing time attribute.
protected void	setRowtimeAttributeDescriptors(List<org.apache.flink.table.sources.RowtimeAttributeDescriptor> rowtimeAttributeDescriptors) Declares a list of fields to be rowtime attributes.
protected void	setSpecificStartupOffsets(Map<KafkaTopicPartition,Long> specificStartupOffsets) Sets the startup offsets of the TableSource; only relevant when the startup mode is StartupMode.SPECIFIC_OFFSETS.
protected void	setStartupMode(StartupMode startupMode) Sets the startup mode of the TableSource.

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaTableSource

protected KafkaTableSource(String topic,
                           Properties properties,
                           org.apache.flink.table.api.TableSchema schema,
                           TypeInformation<Row> returnType)

Creates a generic Kafka StreamTableSource.

Parameters:

topic - Kafka topic to consume.

properties - Properties for the Kafka consumer.

schema - Schema of the produced table.

returnType - Type information of the produced physical DataStream.

Method Detail

getDataStream

public DataStream<Row> getDataStream(StreamExecutionEnvironment env)

NOTE: This method is for internal use only for defining a TableSource. Do not use it in Table API programs.

Specified by:

getDataStream in interface org.apache.flink.table.sources.StreamTableSource<Row>

getReturnType

public TypeInformation<Row> getReturnType()

Specified by:

getReturnType in interface org.apache.flink.table.sources.TableSource<Row>

getTableSchema

public org.apache.flink.table.api.TableSchema getTableSchema()

Specified by:

getTableSchema in interface org.apache.flink.table.sources.TableSource<Row>

getProctimeAttribute

public String getProctimeAttribute()

Specified by:

getProctimeAttribute in interface org.apache.flink.table.sources.DefinedProctimeAttribute

getRowtimeAttributeDescriptors

public List<org.apache.flink.table.sources.RowtimeAttributeDescriptor> getRowtimeAttributeDescriptors()

Specified by:

getRowtimeAttributeDescriptors in interface org.apache.flink.table.sources.DefinedRowtimeAttributes

explainSource

public String explainSource()

Specified by:

explainSource in interface org.apache.flink.table.sources.TableSource<Row>

equals

public boolean equals(Object o)

Overrides:

equals in class Object

hashCode

public int hashCode()

Overrides:

hashCode in class Object

getKafkaConsumer

protected FlinkKafkaConsumerBase<Row> getKafkaConsumer(String topic,
                                                       Properties properties,
                                                       DeserializationSchema<Row> deserializationSchema)

Returns a version-specific Kafka consumer with the start position configured.

Parameters:

topic - Kafka topic to consume.

properties - Properties for the Kafka consumer.

deserializationSchema - Deserialization schema to use for Kafka records.

Returns:

The version-specific Kafka consumer

setProctimeAttribute

protected void setProctimeAttribute(String proctimeAttribute)

Declares a field of the schema to be the processing time attribute.

Parameters:

proctimeAttribute - The name of the field that becomes the processing time field.

setRowtimeAttributeDescriptors

protected void setRowtimeAttributeDescriptors(List<org.apache.flink.table.sources.RowtimeAttributeDescriptor> rowtimeAttributeDescriptors)

Declares a list of fields to be rowtime attributes.

Parameters:

rowtimeAttributeDescriptors - The descriptors of the rowtime attributes.

setStartupMode

protected void setStartupMode(StartupMode startupMode)

Sets the startup mode of the TableSource.

Parameters:

startupMode - The startup mode.

setSpecificStartupOffsets

protected void setSpecificStartupOffsets(Map<KafkaTopicPartition,Long> specificStartupOffsets)

Sets the startup offsets of the TableSource; only relevant when the startup mode is StartupMode.SPECIFIC_OFFSETS.

Parameters:

specificStartupOffsets - The startup offsets for different partitions.

createKafkaConsumer

protected abstract FlinkKafkaConsumerBase<Row> createKafkaConsumer(String topic,
                                                                   Properties properties,
                                                                   DeserializationSchema<Row> deserializationSchema)

Creates a version-specific Kafka consumer.

Parameters:

topic - Kafka topic to consume.

properties - Properties for the Kafka consumer.

deserializationSchema - Deserialization schema to use for Kafka records.

Returns:

The version-specific Kafka consumer

getDeserializationSchema

protected abstract DeserializationSchema<Row> getDeserializationSchema()

Returns the deserialization schema.

Returns:

The deserialization schema