Java写入ClickHouse

╰半橙微兮° 2024-03-26 13:34 94阅读 0赞

#### 文章目录 ####

*  一、添加依赖
 *  二、sink实现
 *   *  1、实体类
     *  2、sink代码

## 一、添加依赖 ##

在pom.xml 增加clickhouse 客户端依赖

<dependency>
        <groupId>ru.yandex.clickhouse</groupId>
        <artifactId>clickhouse-jdbc</artifactId>
        <version>0.2.4</version>
    </dependency>

## 二、sink实现 ##

### 1、实体类 ###

package bean;
    
    public class Person {
        
        public int id;
        public String name;
        public int age;
    
        public Person(int id, String name, int age) {
        
            this.id = id;
            this.name = name;
            this.age = age;
        }
    }

### 2、sink代码 ###

import bean.Person;
    import ru.yandex.clickhouse.BalancedClickhouseDataSource;
    import ru.yandex.clickhouse.ClickHouseConnection;
    import ru.yandex.clickhouse.settings.ClickHouseProperties;
    
    import java.lang.reflect.Field;
    import java.sql.PreparedStatement;
    import java.sql.SQLException;
    
    public class ClickHouseSink{
        
    
        private static final ClickHouseProperties properties = new ClickHouseProperties();
        private static BalancedClickhouseDataSource ds;
        private static ClickHouseConnection connection;
        private static PreparedStatement ps;
        //插入语句
        private static final String SQL = "insert into test.person(id,name,age) values(?,?,?)";
    
        /**
         * 初始化配置
         */
        private static void initialize() {
        
            try {
        
                properties.setUser("distributed-write");
                properties.setPassword("CloudMinds#");
                ds = new BalancedClickhouseDataSource("jdbc:clickhouse://172.16.32.96:9090/ceph_meta", properties);
                connection = ds.getConnection();
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }
    
        /**
         * 关闭连接
         */
        private static void close() {
        
            if (ps != null) {
        
                try {
        
                    ps.close();
                } catch (SQLException e) {
        
                    e.printStackTrace();
                }
            }
            if (connection != null) {
        
                try {
        
                    connection.close();
                } catch (SQLException e) {
        
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] args) {
        
            initialize();
            Person person = new Person(10,"王五",15);
            insertData(SQL,person);
        }
    
        /**
         * 插入数据到clickhouse
         */
        public static <T> void insertData(String sql,T line ) {
        
            try {
        
                ps = connection.prepareStatement(sql);
                Field[] fields = line.getClass().getDeclaredFields();
                for (int i = 0; i < fields.length; i++) {
        
                    Field field = fields[i];
                    try {
        
                        Object o = field.get(line);
                        ps.setObject(i + 1, o);
                    } catch (IllegalAccessException e) {
        
                        e.printStackTrace();
                    }
                }
                ps.addBatch();
                ps.executeBatch();
                ps.clearBatch();
                close();
                connection.close();
            } catch (SQLException e) {
        
                e.printStackTrace();
            }
        }