flinksql写doris

Flink SQL 可以通过 JDBC 连接 Doris 数据库进行数据读写操作。以下是一个简单的 Flink SQL 示例，展示如何使用 Flink SQL 将数据写入 Doris 数据库：

-- 创建一个 Kafka 数据源
CREATE TABLE source_table (
  id INT,
  name STRING,
  age INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'topic' = 'input_topic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'flink_consumer_group',
  'format' = 'json'
);

-- 将数据写入 Doris 数据库
CREATE TABLE sink_table (
  id INT,
  name STRING,
  age INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://localhost:9030/dorisdb',
  'table-name' = 'sink_table',
  'username' = 'root',
  'password' = 'root',
  'driver' = 'com.mysql.jdbc.Driver'
);

-- 将数据从 Kafka 数据源写入 Doris 数据库
INSERT INTO sink_table
SELECT id, name, age FROM source_table;

在上面的示例中，我们首先创建了一个 Kafka 数据源表 `source_table`，然后创建了一个 Doris 数据库表 `sink_table`，最后使用 `INSERT INTO` 语句将数据从 `source_table` 写入到 `sink_table`。需要注意的是，Doris 数据库的 JDBC 连接信息需要根据实际情况进行修改。