org.apache.flink.streaming.connectors.kafka

Class KafkaTableSourceBase

java.lang.Object

org.apache.flink.streaming.connectors.kafka.KafkaTableSourceBase

All Implemented Interfaces:

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

Direct Known Subclasses:

Kafka010TableSource, Kafka011TableSource, Kafka08TableSource, Kafka09TableSource, KafkaTableSource

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

A version-agnostic Kafka StreamTableSource.

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

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	KafkaTableSourceBase(TableSchema schema, Optional<String> proctimeAttribute, List<org.apache.flink.table.sources.RowtimeAttributeDescriptor> rowtimeAttributeDescriptors, Optional<Map<String,String>> fieldMapping, String topic, Properties properties, DeserializationSchema<Row> deserializationSchema, StartupMode startupMode, Map<KafkaTopicPartition,Long> specificStartupOffsets) Creates a generic Kafka StreamTableSource.
protected	KafkaTableSourceBase(TableSchema schema, String topic, Properties properties, DeserializationSchema<Row> deserializationSchema) 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() Describes the table source.
DataStream<Row>	getDataStream(StreamExecutionEnvironment env) NOTE: This method is for internal use only for defining a TableSource.
DeserializationSchema<Row>	getDeserializationSchema() Returns the deserialization schema.
Map<String,String>	getFieldMapping() Returns the mapping for the fields of the TableSource's TableSchema to the fields of its return type TypeInformation.
protected FlinkKafkaConsumerBase<Row>	getKafkaConsumer(String topic, Properties properties, DeserializationSchema<Row> deserializationSchema) Returns a version-specific Kafka consumer with the start position configured.
String	getProctimeAttribute()
Properties	getProperties() Returns the properties for the Kafka consumer.
TypeInformation<Row>	getReturnType() Returns the TypeInformation for the return type of the TableSource.
List<org.apache.flink.table.sources.RowtimeAttributeDescriptor>	getRowtimeAttributeDescriptors()
TableSchema	getTableSchema() Returns the schema of the produced table.
int	hashCode()

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaTableSourceBase

protected KafkaTableSourceBase(TableSchema schema,
                               Optional<String> proctimeAttribute,
                               List<org.apache.flink.table.sources.RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
                               Optional<Map<String,String>> fieldMapping,
                               String topic,
                               Properties properties,
                               DeserializationSchema<Row> deserializationSchema,
                               StartupMode startupMode,
                               Map<KafkaTopicPartition,Long> specificStartupOffsets)

Creates a generic Kafka StreamTableSource.

Parameters:

schema - Schema of the produced table.

proctimeAttribute - Field name of the processing time attribute.

rowtimeAttributeDescriptors - Descriptor for a rowtime attribute

fieldMapping - Mapping for the fields of the table schema to fields of the physical returned type.

topic - Kafka topic to consume.

properties - Properties for the Kafka consumer.

deserializationSchema - Deserialization schema for decoding records from Kafka.

startupMode - Startup mode for the contained consumer.

specificStartupOffsets - Specific startup offsets; only relevant when startup mode is StartupMode.SPECIFIC_OFFSETS.

KafkaTableSourceBase

protected KafkaTableSourceBase(TableSchema schema,
                               String topic,
                               Properties properties,
                               DeserializationSchema<Row> deserializationSchema)

Creates a generic Kafka StreamTableSource.

Parameters:

schema - Schema of the produced table.

topic - Kafka topic to consume.

properties - Properties for the Kafka consumer.

deserializationSchema - Deserialization schema for decoding records from Kafka.

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()

Description copied from interface: TableSource

Returns the TypeInformation for the return type of the TableSource. The fields of the return type are mapped to the table schema based on their name.

Specified by:

getReturnType in interface TableSource<Row>

Returns:

The type of the returned DataSet or DataStream.

getTableSchema

public TableSchema getTableSchema()

Description copied from interface: TableSource

Returns the schema of the produced table.

Specified by:

getTableSchema in interface TableSource<Row>

Returns:

The TableSchema of the produced table.

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

getFieldMapping

public Map<String,String> getFieldMapping()

Description copied from interface: DefinedFieldMapping

Returns the mapping for the fields of the TableSource's TableSchema to the fields of its return type TypeInformation.

The mapping is done based on field names, e.g., a mapping "name" -> "f1" maps the schema field "name" to the field "f1" of the return type, for example in this case the second field of a Tuple.

The returned mapping must map all fields (except proctime and rowtime fields) to the return type. It can also provide a mapping for fields which are not in the TableSchema to make fields in the physical TypeInformation accessible for a TimestampExtractor.

Specified by:

getFieldMapping in interface DefinedFieldMapping

Returns:

A mapping from TableSchema fields to TypeInformation fields or null if no mapping is necessary.

explainSource

public String explainSource()

Description copied from interface: TableSource

Describes the table source.

Specified by:

explainSource in interface TableSource<Row>

Returns:

A String explaining the TableSource.

getProperties

public Properties getProperties()

Returns the properties for the Kafka consumer.

Returns:

properties for the Kafka consumer.

getDeserializationSchema

public DeserializationSchema<Row> getDeserializationSchema()

Returns the deserialization schema.

Returns:

The deserialization schema

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

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