Flink的Sink支持mysql和Clickhouse

一开始确实引入了很多开源的flink-sink-Clickhouse的包但是还是不理想高并发的情况下偶尔会丢数据或者job直接卡死的情况，所以最后使用官方的jdbc来连接找了一波在站内看到大佬的帖子。里面提到的开源项目都做过尝试，最后在博客园翻到一个博主提供了jdbc的示例，于是在他的基础上改了一波。其他的基本操作都是一样的，都是通过PreparedStatement来指定构造sql的参数，一开

莫小牙阝

213人浏览 · 2023-09-08 16:19:35

莫小牙阝 · 2023-09-08 16:19:35 发布

最近遇到一个问题需要把flink的数据直接输入到Clickhouse中，一开始直接输入到kafka然后通过视图去监听topic在入库，但是中间延迟严重以及会丢数据的情况，所以引入需要直接输入到Clickhouse库中。当前项目中有基于jdbc的还有基于中间件模式的，具体中间件模式的使用可以查看站内链接的博客。

#官方

JDBC | Apache Flink

#csdn

flink-connector-clickhouse_flink clickhouse connector-CSDN博客

#博客园

https://www.cnblogs.com/sqhhh/p/15897275.html

最后基于官方jdbc的驱动重新封装直接丢上去跑了一波感觉还行，就此分享记录下。

flink环境为

 <scala.version>2.11</scala.version>
 <flink.version>1.14.4</flink.version>

相关jdbc驱动

 <!-- ClickHouse JDBC 驱动依赖 -->
        <dependency>
            <groupId>ru.yandex.clickhouse</groupId>
            <artifactId>clickhouse-jdbc</artifactId>
            <version>0.3.1</version>
        </dependency>

        <!-- mysql JDBC 驱动依赖 -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.29</version>
        </dependency>

    public SinkSingleDB(String sql, JdbcStatementBuilder<T> jdbcStatementBuilder,
                        Properties props) {
        sink = JdbcSink.sink(
                sql,
                jdbcStatementBuilder,
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:" + props.getProperty(Constants.DRIVE) + "://" + props.getProperty(Constants.HOST) + "/" + props.getProperty(Constants.DATABASE))
                        .withDriverName(SinkSourceEnums.getDriverByType(props.getProperty(Constants.DRIVE)))
                        .withUsername(props.getProperty(Constants.USERNAME))
                        .withPassword(props.getProperty(Constants.PASSWORD))
                        .build()
        );
    }

public SinkSingleDB(String sql, JdbcStatementBuilder<T> jdbcStatementBuilder,
                        Properties props, Boolean options) {
        sink = JdbcSink.sink(
                sql,
                jdbcStatementBuilder,
                JdbcExecutionOptions.builder()
                        .withBatchIntervalMs(Long.valueOf(props.getProperty(Constants.BATCH_INTERVAL_MS, DEFAULT_INTERVAL_MILLIS)))
                        .withBatchSize(Integer.valueOf(props.getProperty(Constants.BATCH_SIZE, DEFAULT_SIZE)))
                        .withMaxRetries(Integer.valueOf(props.getProperty(Constants.MAX_RETRIES, DEFAULT_MAX_RETRY_TIMES)))
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:" + props.getProperty(Constants.DRIVE) + "://" + props.getProperty(Constants.HOST) + "/" + props.getProperty(Constants.DATABASE))
                        .withDriverName(SinkSourceEnums.getDriverByType(props.getProperty(Constants.DRIVE)))
                        .withUsername(props.getProperty(Constants.USERNAME))
                        .withPassword(props.getProperty(Constants.PASSWORD))
                        .build()
        );
    }

当前值连接jdbc的老手应该很熟悉，唯一区别就是第二指定了批处理可以按照自己的性能要求直接提交次数和数量。

如果不指定就是默认时间。时间间隔和数据量满足一个则提交，需要注意的是时间间隔为0是不限，可以查看官方文档