flink连接doris

Apache Flink 可以通过 JDBC 连接器连接到 Doris，以实现将 Flink 数据流写入 Doris 表的目标。下面是 Flink 连接 Doris 的步骤：

1. 首先，需要将 Doris JDBC 驱动程序添加到 Flink 项目的依赖项中。可以在 Maven 中添加以下依赖项：

<dependency>
  <groupId>org.apache.calcite.avatica</groupId>
  <artifactId>avatica-core</artifactId>
  <version>1.13.0</version>
</dependency>
<dependency>
  <groupId>org.apache.doris</groupId>
  <artifactId>doris-jdbc</artifactId>
  <version>0.14.0</version>
</dependency>

2. 在 Flink 程序中创建一个 JDBC 连接，并将其用于写入 Doris。下面是一个示例代码片段：

// 创建 Doris 的 JDBC 连接
Properties properties = new Properties();
properties.put("user", "root");
properties.put("password", "root");
Connection dorisConnection = DriverManager.getConnection("jdbc:doris://localhost:9030/test", properties);

// 将 Flink 数据流写入 Doris 表
DataStream<Tuple2<String, Integer>> dataStream = ...;
dataStream.addSink(JdbcSink.sink(
  "INSERT INTO `test`.`word_count` (`word`, `count`) VALUES (?, ?)",
  (ps, t) -> {
    ps.setString(1, t.f0);
    ps.setInt(2, t.f1);
  },
  new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
    .withUrl("jdbc:doris://localhost:9030/test")
    .withUsername("root")
    .withPassword("root")
    .build(),
  new JdbcExecutionOptions.JdbcExecutionOptionsBuilder()
    .withBatchSize(100)
    .build()
));

在这个示例中，我们使用 `JdbcSink` 将 Flink 数据流写入 Doris 表。`JdbcSink` 接受 SQL 查询字符串、参数设置函数和 JDBC 连接选项作为参数。在参数设置函数中，我们将 Flink 数据流中的元组转换为 SQL 参数。`JdbcConnectionOptions` 和 `JdbcExecutionOptions` 用于配置 JDBC 连接选项和执行选项。

3. 在 Doris 中创建一个表，以存储 Flink 数据流。下面是一个示例 SQL 语句：

CREATE TABLE `test`.`word_count` (
  `word` varchar(255) NOT NULL,
  `count` int(11) DEFAULT NULL,
  PRIMARY KEY (`word`)
) ENGINE=OLAP

这个示例创建了一个名为 `word_count` 的表，其中包含两个列：`word` 和 `count`。`word` 列是主键列。

通过以上步骤，我们就可以将 Flink 数据流写入 Doris 表了。