Kafka Connector Introduction

GeaFlow support read data from kafka and write data to kafka. Currently support kafka version is 2.4.1.

Syntax

CREATE TABLE kafka_table (
  id BIGINT,
  name VARCHAR,
  age INT
) WITH (
	type='kafka',
    geaflow.dsl.kafka.servers = 'localhost:9092',
	geaflow.dsl.kafka.topic = 'test-topic'
)

Options

Key	Required	Description
geaflow.dsl.kafka.servers	true	The kafka bootstrap servers list.
geaflow.dsl.kafka.topic	true	The kafka topic.
geaflow.dsl.kafka.group.id	false	The kafka group id. Default value is: 'default-group-id'.

Example

CREATE TABLE kafka_source (
  id BIGINT,
  name VARCHAR,
  age INT
) WITH (
	type='kafka',
    geaflow.dsl.kafka.servers = 'localhost:9092',
	geaflow.dsl.kafka.topic = 'read-topic'
);

CREATE TABLE kafka_sink (
  id BIGINT,
  name VARCHAR,
  age INT
) WITH (
	type='kafka',
    geaflow.dsl.kafka.servers = 'localhost:9092',
	geaflow.dsl.kafka.topic = 'write-topic'
);

INSERT INTO kafka_sink
SELECT * FROM kafka_source;