Kafka同步到Doris
本文提供Kafka同步Doris场景的实施思路、配置示例和验证方式，可作为实际项目配置参考。
案例
将kafka(3.1.2)同步到doris(2.0.13)数据库中
解决方案
- kafka-clients-3.1.2.jar下载
- flink-doris-connector-1.18-25.0.0.jar下载
- flink-JSON-1.18.1.jar下载
- flink-connector-kafka-3.2.0-1.18.jar下载
示例sql如下：
CREATE TABLE from_table
(
username string,
age int
) WITH (
'connector' = 'kafka',
'topic' = 'ispong-topic',
'properties.bootstrap.servers' = '127.0.0.1:9092',
'scan.startup.mode' = 'latest-offset',
'format' = 'json',
'json.fail-on-missing-field' = 'false',
'json.ignore-parse-errors' = 'true'
);
CREATE TABLE target_table
(
username string,
age int
) WITH (
'connector' = 'doris',
'fenodes' = '127.0.0.1:9083',
'table.identifier' = 'ispong_db.users',
'username' = 'admin',
'password' = '',
'sink.enable.batch-mode' = 'true',
'sink.buffer-flush.max-bytes' = '50MB'
);
INSERT INTO target_table (username, age)
select username, age
from from_table;