flink从kafka读取数据写入clickhouse本地表的实现

因为直接写clickhouse的分布式表在数据量比较大的时候会有各种问题，所以做了一个flink读取kafka数据然后路由写入到相应的本地表节点，并且关于不同的表的配置信息可以随时更改并设置生效时间。

首先从kafka将数据读取过来然后进行相应的处理及逻辑判断写入到对应的clickhouse表格中最后根据CDC读取来的配置信息进行相应节点的hash路由，直接写入本地表

定义kafka工具类，实现kafka的source获取函数

public static FlinkKafkaConsumer getKafkaSource(String topic, String groupId) {
        Properties prop = new Properties();
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        return new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), prop);
    }

之后在主程序中获取环境，并读取数据

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


env.addSource(KafkaUtils.getKafkaSource(topic,groupId).setStartFromGroupOffsets()).setParallelism(4)

这里将clickhouse的分布式表的配置信息存储在MySQL中，利用CDC进行一个实时的监控和更改

在配置流没有到达之前，将采用代码中的默认配置进行处理，同时将默认配置的数据存储在Redis中，也可以做到很方便的修改。

//flinkcdc 读取配置
DebeziumSourceFunction configSource = MySQLSource.builder()
       .hostname("******")
       .port(3306)
       .databaseList("test_config")
       .tableList("test_config.ck_config_info")
       .username("root")
       .password("******")
       .startupOptions(StartupOptions.initial())
       .deserializer(new MyCustomSchema())
       .build();

其中的序列化采用自定义方式，会对每一条数据打一个标记，是新增还是更新或者删除，以便于后续配置状态的更新

配置流准备好后开始和主流进行广播连接，之后自定义处理函数继承BroadcastProcessFunction，来进行数据的节点路由

主要是将配置流数据存进一个状态里，然后主流数据每次都去状态里读取自己需要的配置信息，然后进行想要写入的本地表节点的路由写入

主流数据路由逻辑如下

if(tableConfig != null){
            String ipList = tableConfig.getIpList();
            String[] split = ipList.split(",");

            assert stringRowTuple2.f1.getField(0) != null;

            //hash获取对应节点
            int index = CommonUtils.getHash(stringRowTuple2.f1.getField(0).toString()) % split.length;
            String ip = split[index];
            collector.collect(Tuple2.of(ip,stringRowTuple2.f1));
        }else {
            System.out.println("no this key in TableConfig:" + stringRowTuple2.f0);
            System.out.println("use default......");
            collector.collect(Tuple2.of("default",stringRowTuple2.f1));
        }

对于采用default方案的数据，会在后续处理中采用默认的节点进行相同逻辑的路由，同时，这要求我们在进行row类型数据存放的时候，需要将主键放在第一位，以达到相同主键的数据路由到同一节点，以便于后续根据不同表引擎的特性进行去重或者其他处理。

最关键的就是sink函数的编写，这里可以参考flink的JdbcSink.sink()的实现进行处理

public class ClickHouseJDBCSinkFunction extends RichSinkFunction> {

    final ClickHouseJDBCOutputFormat clickHouseJDBCOutputFormat;

    public ClickHouseJDBCSinkFunction(ClickHouseJDBCOutputFormat clickHouseJDBCOutputFormat) {
        this.clickHouseJDBCOutputFormat = clickHouseJDBCOutputFormat;
    }



    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        RuntimeContext ctx = getRuntimeContext();
        clickHouseJDBCOutputFormat.setRuntimeContext(ctx);
        clickHouseJDBCOutputFormat.open(ctx.getIndexOfThisSubtask(), ctx.getNumberOfParallelSubtasks());
        System.out.println("init...");
    }

    @Override
    public void invoke(Tuple2 value, Context context) throws Exception {
        clickHouseJDBCOutputFormat.writeRecord(value);
    }


    @Override
    public void close() throws Exception {
        clickHouseJDBCOutputFormat.close();
        super.close();
    }
}

写入表的操作在writeRecord方法里实现

进入此方法会根据设定的时间间隔及批次大小决定是将批次数据刷写入表还是继续往批次里面添加数据，当达到刷写时机的时候会调用flush()方法来进行书写，并且针对刷写失败的情况设置了重试次数

private void flush(List rows, PreparedStatement preparedStatement, Connection connection,int maxRetryTimes,String ip) throws SQLException {
        checkFlushException();

        if(maxRetryTimes>0){

            DateUtils.sleepMinutes(CommonUtils.MAX_RETRY_TIMES-maxRetryTimes);

            try {
                for (int i = 0; i < rows.size(); ++i) {
                    Row row = rows.get(i);
                    for (int j = 0; j < this.tableColumns.length; ++j) {

                        if (null != row.getField(j)) {
                            preparedStatement.setObject(j + 1, row.getField(j));
                        } else {
                            preparedStatement.setObject(j + 1, "null");
                        }
                    }
                    preparedStatement.addBatch();
                }

                preparedStatement.executeBatch();
                connection.commit();
                preparedStatement.clearBatch();

                rows.clear();

             
            } catch (Exception e){
                System.out.println("failed to insert ,will try it "+(CommonUtils.MAX_RETRY_TIMES-maxRetryTimes+1)+" minutes again......");

                //判断连接是否可用，不可用重新获取连接
                if(!connection.isValid(60)){
                    try {
                        reEstablishConnection(ip);

                        preparedStatement = connection.prepareStatement(query);
                        this.preparedStatementMap.put(ip,preparedStatement);


                    } catch (SQLException sqe) {
                        throw new IllegalArgumentException("open() failed.", sqe);
                    } catch (ClassNotFoundException cnfe) {
                        throw new IllegalArgumentException("JDBC driver class not found.", cnfe);
                    }
                }

                flush(rows, preparedStatement, connection, maxRetryTimes-1,ip);
            }

        }else {
            //三次重试失败后换一个节点继续尝试发送
            String otherIp = findOtherIp(ip);
            flush(rows,preparedStatementMap.get(otherIp), connectionMap.get(otherIp),CommonUtils.MAX_RETRY_TIMES,otherIp);
        }

    }

这里我设置的是三次重试失败后会换一个节点继续重试，然后一直循环，除非没有了可用节点，当然也可以不这么做，我们也能够将写入失败的数据写入临时文件或者采用其他方式

对于当批次数量达不到的情况下，在时间间隔内触发flush的设置，我这边尝试了两种方案，一种是采用Java的scheduledFuture来进行设置，但是这种方案会有个问题，就是在一个时间间隔的末尾阶段正好批次量达到的时候，会在短时间内进行两次刷写操作，第二次写入的数据量可能会很小，这在使用的时候需要注意。另一种方案就是采用系统时间来进行设置，通过不断更新上次刷写的系统时间并与当前系统时间进行做差值，当差值满足条件时进行刷写操作，这可以避免上个方案的问题，但是这要求数据流不能断，因为当数据流没数据后，就无法在对时间进行作差，会导致最后一批数据迟迟不能写入表中。

在数据量不大的情况下写入数据可以直接写入分布式表，这无可厚非，毕竟代码实现起来很方便，这次是对于写本地表的一次尝试练习，经过测试，想要的功能基本实现，以此记录，对于其中的不足之处也望大家多多指正。