Clickhouse在流式数据中的批量写入

使用场景
- 实现思路
- 代码实现
- - 结语

使用场景

对于clickhouse有过使用经验的开发者应该知道，ck的写入，最优应该是批量的写入，但是对于流式场景来说，每批写入的数据量都是不可控制的，如kafka，每批拉取的消息数量是不定的，flink对于每条数据流的输出，写入ck的效率会十分缓慢，所以写了一个demo,去批量入库，生产环境使用还需要优化

实现思路

我的思路是，维护一个缓存队列当做一个缓冲区，当队列数据条数到达一定阈值，或者数据滞留时间超过一定时间，此时进行ck的批量提交。

代码实现

flink 写入clickhouse 的sink

package com.su.data.sink;

import com.su.data.pojo.RouteInfoPoJO;
import com.su.data.util.ClickhouseTasK;
import com.su.data.util.ClickhouseUtil;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;


import java.io.Serializable;




public class ClickhouseSink extends RichSinkFunction implements  Serializable {


    String sql;


    public ClickhouseSink(String sql) {
        this.sql = sql;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        ClickhouseTasK.connection  = ClickhouseUtil.getConn();
    }

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

    @Override
    public void invoke(RouteInfoPoJO routeInfoPoJO, Context context) throws Exception {
    	//流式数据写入缓存
        ClickhouseTasK.getInstance(sql).totalAdd(routeInfoPoJO);
    }
}

数据处理模块

package com.su.data.util;

import com.su.data.pojo.RouteInfoPoJO;
import com.su.data.sink.ClickhouseSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;



public class ClickhouseTasK {

    
    private static final int MAXSIZE = 10000;

    
    private static Queue queue = new LinkedBlockingQueue<>(MAXSIZE);

    
    private static Queue TOTAL_QUEUE = new ConcurrentLinkedQueue();

    
    private static ReentrantLock queenLock =null;

    
    private static volatile ClickhouseTasK clickhouseTasK = null;

    
    private static Condition FULL = null;

    
    private static Condition UN_FULL = null;
    
    public static Connection connection = null;


    
    private static final long MAX_WAIT_TIME = 3000;

    
    private  static Thread dataThread = null;

    
    private  static Thread moveThread = null;

    
    private static Thread timeThread = null;

    
    static  AtomicLong atomicLong = null;

    
    static AtomicLong count = null;

    
    static {
        count = new AtomicLong(0);
        //有界队列  10000条提交一次
        queenLock = new ReentrantLock();
        FULL = queenLock.newCondition();
        UN_FULL =  queenLock.newCondition();
    }

    
    public static ClickhouseTasK getInstance(String sql) throws  InterruptedException {
        if( null == clickhouseTasK){
            synchronized (ClickhouseTasK.class){
                if(null == clickhouseTasK){
                    clickhouseTasK = new ClickhouseTasK(sql);
                }
            }
        }
        return  clickhouseTasK;
    }

    
    public ClickhouseTasK(String sql) throws  InterruptedException {
        atomicLong  = new AtomicLong(System.currentTimeMillis());
        CountDownLatch countDownLatch = new CountDownLatch(2);
        //时间记录线程
        timeThread = new Thread(()->{
            while (true){
                queenLock.lock();
                try {
                //时间一到 自动提交
                if(System.currentTimeMillis() - atomicLong.get()  >= MAX_WAIT_TIME && queue.size() >0 ){
                        System.out.println("到时间自动提交："+queue.size());
                        //数据提交
                        commitData(queue,connection,sql);
                        //记录本次提交的时间
                        atomicLong.set(System.currentTimeMillis());
                        // 数据提交后，提交队列空了  唤醒 阻塞于UN_FULL的线程
                        UN_FULL.signal();
                }
                } catch (SQLException e) {

                    System.out.println("中断异常："+e);
                }finally {
                    queenLock.unlock();
                }
            }
        });
        timeThread.setName("timeThread");
        //数据拉去线程
        moveThread = new Thread(() ->{
            countDownLatch.countDown();
            while (true){
                if(!TOTAL_QUEUE.isEmpty()){
                    add(TOTAL_QUEUE.poll());
                }
            }
        });
        moveThread.setName("moveThread");
        dataThread = new Thread(() ->{
            System.out.println(Thread.currentThread().getName() + "启动！！！");
            countDownLatch.countDown();
            while (true){
                queenLock.lock();
                try {
                    //等待队列满了 当阻塞于FULL的线程被唤醒，说明队列满了
                    FULL.await();
                    //提交逻辑 清空队列
                    if(!queue.isEmpty()){
                        commitData(queue,connection,sql);
                    }

                    //重置提交时间
                    atomicLong.set(System.currentTimeMillis());
                    // 数据提交后，提交队列空了  唤醒 阻塞于UN_FULL的线程
                    UN_FULL.signal();
                } catch (SQLException | InterruptedException exception) {
                    System.out.println("中断异常："+exception);
                }finally {
                    queenLock.unlock();
                }
            }
        });
        dataThread.setName("dataQueenThread");
        moveThread.start();
        dataThread.start();
        timeThread.start();
        //确保各线程启动完成后 构造函数线程初始化完成
        countDownLatch.await();
        System.out.println("初始化完成了");

    }
    public void commitData(Queue queue, Connection connection, String sql) throws SQLException {
        System.out.println("准备提交，当前数量为："+queue.size());
        //批量提交
        try(PreparedStatement preparedStatement = connection.prepareStatement(sql)){
            long startTime = System.currentTimeMillis();
            while (!queue.isEmpty()){
                RouteInfoPoJO routeInfoPoJO = queue.poll();
                preparedStatement.setString(1, routeInfoPoJO.getDeal_date());
                preparedStatement.setString(2, routeInfoPoJO.getClose_date());
                preparedStatement.setString(3, routeInfoPoJO.getCard_no());
                preparedStatement.setString(4, routeInfoPoJO.getDeal_value());
                preparedStatement.setString(5, routeInfoPoJO.getDeal_type());
                preparedStatement.setString(6, routeInfoPoJO.getCompany_name());
                preparedStatement.setString(7, routeInfoPoJO.getCar_no());
                preparedStatement.setString(8, routeInfoPoJO.getStation());
                preparedStatement.setString(9, routeInfoPoJO.getConn_mark());
                preparedStatement.setString(10, routeInfoPoJO.getDeal_money());
                preparedStatement.setString(11, routeInfoPoJO.getEqu_no());
                preparedStatement.addBatch();
            }
            //ck没有事务，提交了就执行了
            int[] ints = preparedStatement.executeBatch();
            long endTime = System.currentTimeMillis();
            System.out.println("批量插入完毕用时：" + (endTime - startTime) + " -- 插入数据 = " + ints.length);
            System.out.println("现有总量："+count.get());
        }
        // todo 真实场景下，数据要确保不丢失，需要对异常数据进行处理，如日志记录后，进行数据日志采集 重复入库即可
        // todo 想要确保ck数据不重复，建立时选择replacing合并树，然后重复数据自动合并就好了
    }

    public void  add (RouteInfoPoJO routeInfoPoJO){
        //满足 量 提交条件
        queenLock.lock();
        try {
            //提交队列满了
            if (queue.size() >= MAXSIZE ){
                //唤醒阻塞于 FULL的线程
                FULL.signal();
                //阻塞 UN_FULL上的线程
                UN_FULL.await();
            }
            if(routeInfoPoJO !=null){
                //提交队列入队
                queue.offer(routeInfoPoJO);
            }
        } catch (InterruptedException exception) {
            exception.printStackTrace();
        } finally {
            queenLock.unlock();
        }
    }

    
    public void totalAdd(RouteInfoPoJO routeInfoPoJO){
        TOTAL_QUEUE.add(routeInfoPoJO);
        count.incrementAndGet();
    }
}

RouteInfoPoJO 实体

package com.su.data.pojo;

import lombok.AllArgsConstructor;
import lombok.Data;



@AllArgsConstructor
@Data
public class RouteInfoPoJO {
    private String deal_date;
    private String close_date;
    private String card_no;
    private String deal_value;
    private String deal_type;
    private String company_name;
    private String car_no;
    private String station;
    private String conn_mark;
    private String deal_money;
    private String equ_no;
}

flink 从kafka 写入ck

package com.su.data;

import com.alibaba.fastjson.JSONObject;
import com.su.data.pojo.RouteInfoPoJO;
import com.su.data.serializer.MyDeserialization;
import com.su.data.sink.ClickhouseSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;



public class KakfaToCK {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);

        Properties properties = new Properties();
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG,"ckcomsumer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"ck-node");
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"xxxx:9092");


        DataStreamSource dataStreamSource = env.addSource(new FlinkKafkaConsumer011("szt",new MyDeserialization(),properties));

        String sql = "INSERT INTO sz.ods_szt_data(deal_date, close_date, card_no, deal_value, deal_type, company_name, car_no, station, conn_mark, deal_money, equ_no) values (?,?,?,?,?,?,?,?,?,?,?)";
        //dataStreamSource.print();
        dataStreamSource.addSink(new ClickhouseSink(sql));
        env.execute("KakfaToCK");

    }
}

结语

最后，本案例是在学习github上深圳通大数据的项目写的，如果有兴趣可以去看看。
https://gitee.com/geekyouth/SZT-bigdata
如果有写的不对的地方还请大家多多指教！

Clickhouse在流式数据中的批量写入

Java相关栏目本月热门文章