A Flink Kafka connector links a Flink dataflow to Kafka, letting Flink read topic data, process it continuously, and write results back to Kafka. In an online product-price pipeline, the connector can expose price changes to Flink SQL or downstream consumers. section 12.2.1 section 12.3
Configure the connection around these decisions:
- Declare Kafka as the connector and provide the bootstrap server, topic, and data format. section 12.3
- Map message fields and keys explicitly when the Kafka record key carries business identity such as product_id. section 12.3
- Treat Flink SQL queries as continuous operations rather than one-time database queries; write results back to Kafka when consumers need them. section 12.3
Product-price change stream
Create a Kafka-backed table for products.prices.changelog, filter rows where price exceeds 100, or continuously count price changes grouped by product_id. The same connector pattern supports sending processed results to another Kafka topic. section 12.2.1 section 12.3
CREATE TABLE productprices ( product_id INT, product_name STRING, price FLOAT, oldPrice FLOAT ) WITH ( 'connector' = 'kafka', 'topic' = 'products.prices.changelog', 'properties.bootstrap.servers' = 'localhost:9092', 'format' = 'json', 'key.format' = 'json', 'key.fields' = 'product_id', 'value.fields-include' = 'EXCEPT_KEY' );
Inspect the complete Flink SQL CREATE TABLE definition, including the Kafka connector, topic, bootstrap server, JSON format, and key-field settings. section 12.3
