【自定义Source、Sink】Flink自定义Source、Sink对ClickHouse进行读和批量写操作

ClickHouse官网文档

Flink 读取 ClickHouse 数据两种驱动

ClickHouse 官方提供Clickhouse JDBC.【建议使用】
第3方提供的Clickhouse JDBC. ru.yandex.clickhouse.ClickHouseDriver

ru.yandex.clickhouse.ClickHouseDriver.现在是没有维护

ClickHouse 官方提供Clickhouse JDBC的包名：com.clickhouse.jdbc.*

有些版本com.clickhouse.jdbc.* 包含了 ru.yandex.clickhouse.ClickHouseDriver.

因此加载包的时候一定要注意导入的包名

引入依赖

        <!-- clickhouse jdbc driver -->
        <dependency>
            <groupId>com.clickhouse</groupId>
            <artifactId>clickhouse-jdbc</artifactId>
        </dependency>

使用的是 0.3 这个版本，该版本就包含上述3方CH jdbc包

     <!-- CH JDBC版本推荐使用 0.3, 0.4的版本是要 JDK 17 -->
     <clickhouse-jdbc.version>0.3.2-patch11</clickhouse-jdbc.version>

自定义Source

测试表映射实体类，该表仅有一个name字段

@Data
@NoArgsConstructor
@AllArgsConstructor
public class CHTestPO {

    private String name;

}

Flink Clickhouse Source

public class ClickHouseSource implements SourceFunction<CHTestPO> {

    private final String URL;
    private final String SQL;

    public ClickHouseSource(String URL, String SQL) {
        this.URL = URL;
        this.SQL = SQL;
    }

    @Override
    public void run(SourceContext<CHTestPO> output) throws Exception {
        
        //  Properties是持久化的属性集 Properties的key和value都是字符串
        Properties properties = new Properties();
        ClickHouseDataSource clickHouseDataSource = new ClickHouseDataSource(URL, properties);

        // 使用 try-with-resource 方式关闭JDBC连接 无需手动关闭
        try (ClickHouseConnection conn = clickHouseDataSource.getConnection()) {

            // clickhouse 通过游标的方式读取数据
            Statement stmt = conn.createStatement();
            ResultSet rs = stmt.executeQuery(SQL);
            while (rs.next()) {
                String name = rs.getString(1);
                output.collect(new CHTestPO(name));
            }
        }

    }

    @Override
    public void cancel() {

    }
}

自定义Sink

需额外引入依赖

        <!-- Flink-Connector-Jdbc -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
        </dependency>

Java 对sql语句处理的两个对象

PreparedStatement对象：能够对预编译之后的sql语句进行处理【SQL 语句预编译：通过占位符'?'实现，可以防止sql注入】
Statement对象：只能对静态的sql语句进行处理

核心代码

/**
 * 使用 Flink-jdbc-connector + 批量写入 + sql语句的预编译 写入 Clickhouse
 */
public class ClickHouseJdbcSink<T> {

    private final SinkFunction<T> sink;
    private final static String NA = "null";

    public ClickHouseJdbcSink(String sql, int batchSize, String url) {

        sink = JdbcSink.sink(
                sql,
                // 对sql语句进行预编译
                new ClickHouseJdbcStatementBuilder<T>(),
                // 设置批量插入数据
                new JdbcExecutionOptions.Builder().withBatchSize(batchSize).build(),
                // 设置ClickHouse连接配置
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(url)
                        .build()
        );

    }

    public SinkFunction<T> getSink() {
        return this.sink;
    }

    /**
     * 对预编译之后的sql语句进行占位符替换
     *
     * @param ps:     PreparedStatement对象 下标从 1 开始
     * @param fields: clickhouse表PO对象的属性字段
     * @param object: clickhouse表PO对象的属性字段所对应的数据类型
     */
    public static void setPreparedStatement(
            PreparedStatement ps,
            Field[] fields,
            Object object) throws IllegalAccessException, SQLException {

        // 遍历 Field[]
        for (int i = 1; i <= fields.length; i++) {
            // 取出每个Field实例
            Field field = fields[i - 1];
            // 指示反射的对象在使用时应该取消 Java 语言访问检查
            field.setAccessible(true);
            // 通过Field实例的get方法返回指定的对象
            Object o = field.get(object);
            if (o == null) {
                ps.setNull(i, 0);
                continue;
            }

            // 这里统一设为字符型
            String fieldValue = o.toString();

            // 变量和常量的比较，通常将常量放前，可以避免空指针
            if (!NA.equals(fieldValue) && !"".equals(fieldValue)) {
                // 替换对应位置的占位符
                ps.setObject(i, fieldValue);
            } else {
                ps.setNull(i, 0);
            }
        }
    }

}

对sql语句进行预编译

@Slf4j
public class ClickHouseJdbcStatementBuilder<T> implements JdbcStatementBuilder<T> {

    @Override
    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {

        /* **********************
         * Java通过反射获取类的字段：
         *
         * 1. getDeclaredFields()：获取所有的字段，不会获取父类的字段
         * 2. getFields(): 只能会public字段，获取包含父类的字段
         *
         * *********************/

        Field[] fields = t.getClass().getDeclaredFields();

        // 将获取到的字段替换sql预编译之后的占位符。
        try {
            ClickHouseJdbcSink.setPreparedStatement(preparedStatement, fields, t);
        } catch (IllegalAccessException e) {
            log.error("sql 预编译失败", e);
            e.printStackTrace();
        }

    }
}

ClickHouse读写工具类

public class ClickHouseUtil {

    private static final String URL;

    static {
        ParameterTool parameterTool = ParameterUtil.getParameters();
        URL = parameterTool.get("clickhouse.url");
    }

    /**
     * 读取clickhouse
     */
    public static DataStream<CHTestPO> read(StreamExecutionEnvironment env, String sql) {
        return env.addSource(new ClickHouseSource(URL, sql));
    }

    /**
     * 批量写入ClickHouse
     */
    public static <T> DataStreamSink<T> batchWrite(
            DataStream<T> dataStream,
            String sql,
            int batchSize) {

        //生成 SinkFunction
        ClickHouseJdbcSink<T> clickHouseJdbcSink =
                new ClickHouseJdbcSink<T>(sql, batchSize, URL);
        return dataStream.addSink(clickHouseJdbcSink.getSink());
    }

}

测试一下

public class ClickHouseUtilTest {

    @DisplayName("测试Flink+jdbc+游标读取Clickhouse")
    @Test
    void testRead() throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度1
        env.setParallelism(1);

        // 从default数据库的user表中读取数据
        String sql = "select * from default.user";
        DataStream<CHTestPO> ds = ClickHouseUtil.read(env, sql);

        // 打印数据流中的元素
        ds.print("clickhouse");

        // 执行程序
        env.execute();

    }

    @DisplayName("测试Flink-Connector-jdbc+预编译批量写入Clickhouse")
    @Test
    void testBatchWrite() throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度1
        env.setParallelism(1);

        CHTestPO po = new CHTestPO();
        po.setName("Lucy");
        CHTestPO po1 = new CHTestPO();
        po1.setName("Jack");

        DataStream<CHTestPO> ds = env.fromCollection(Arrays.asList(po, po1));

        // 定义将数据写入ClickHouse数据库的SQL语句
        String sql = "insert into default.user(name) values(?)";
        // 调用ClickHouseUtil的batchWrite方法将数据流ds中的数据批量写入ClickHouse数据库
        ClickHouseUtil.batchWrite(ds, sql, 2);

        // 执行程序
        env.execute();
    }
}