ApsaraDB for ClickHouse:Connect using JDBC

package com.aliyun;

import com.clickhouse.jdbc.ClickHouseDataSource;
import com.clickhouse.data.ClickHouseOutputStream;
import com.clickhouse.data.ClickHouseWriter;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
    private final static String YOUR_INSTANCE_PROTOCOL = "http";
    private final static String YOUR_INSTANCE_ENDPOINT = "VPC_ENDPOINT:8123"; // YOUR CONFIG HERE
    private final static String DATABASE = "default"; // YOUR CONFIG HERE
    private final static String YOUR_INSTANCE_USER = "USER"; // YOUR CONFIG HERE
    private final static String YOUR_INSTANCE_PASSWORD = "PASSWORD"; // YOUR CONFIG HERE
    private final static String JDBC_URL = "jdbc:clickhouse:%s://%s/%s";
    private final static Integer INSERT_BATCH_SIZE = 10000;
    private final static Integer INSERT_BATCH_NUM = 10;
    private final static boolean ENTERPRISE = true; // YOUR CONFIG HERE
    private final static Integer INSERT_OPTIMIZE_LEVEL = 3;

    public static void main(String[] args) {
        try {
            HikariConfig conf = buildHikariDataSource();
            try(HikariDataSource ds = new HikariDataSource(conf)) {
                // Create a table.
                Connection conn = ds.getConnection();
                createTable(conn);
                conn.close();

                // Concurrently insert data.
                int concurrentNum = 5;
                CountDownLatch countDownLatch = new CountDownLatch(concurrentNum);
                ExecutorService executorService = Executors.newFixedThreadPool(concurrentNum);
                for (int i = 0; i < concurrentNum; i++) {
                    executorService.submit(() -> {
                        System.out.printf("[%d] Thread starts inserting\n", Thread.currentThread().getId());
                        try(Connection connection = ds.getConnection()) {
                            batchInsert(connection, INSERT_OPTIMIZE_LEVEL);
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            System.out.printf("[%d] Thread stops inserting\n", Thread.currentThread().getId());
                            countDownLatch.countDown();
                        }
                    });
                }
                // Wait for all threads to finish.
                countDownLatch.await();

                // Count the table.
                conn = ds.getConnection();
                count(conn);
                conn.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * Generate the JDBC URL.
     * @param protocol The protocol. Supported protocols include http, https, and grpc.
     * @param endpoint The endpoint.
     * @return The JDBC URL.
     */
    public static String getJdbcUrl(String protocol, String endpoint, String database) {
        return String.format(JDBC_URL, protocol, endpoint, database);
    }

    /**
     * Build HikariDataSource.
     * @return The HikariConfig.
     */
    public static HikariConfig buildHikariDataSource() throws Exception {
        HikariConfig conf = new HikariConfig();

        // Properties
        Properties properties = new Properties();
        /// Socket keepalive
        properties.setProperty("socket_keepalive", "true");
        properties.setProperty("http_connection_provider", "APACHE_HTTP_CLIENT");
        /// Socket timeout
        properties.setProperty("socket_timeout", "120000");
        /// Timezone
        properties.setProperty("use_server_time_zone", "true");

        // Data source configuration
        conf.setDataSource(new ClickHouseDataSource(getJdbcUrl(YOUR_INSTANCE_PROTOCOL, YOUR_INSTANCE_ENDPOINT, DATABASE), properties));
        conf.setUsername(YOUR_INSTANCE_USER);
        conf.setPassword(YOUR_INSTANCE_PASSWORD);

        // Connection pool configuration
        conf.setMaximumPoolSize(10);
        conf.setMinimumIdle(5);
        conf.setIdleTimeout(30000);
        conf.setMaxLifetime(60000);
        conf.setConnectionTimeout(30000);
        conf.setPoolName("HikariPool");

        return conf;
    }

    /**
     * Create a table.
     * @param conn The ClickHouse connection.
     * @throws Exception
     */
    public static void createTable(Connection conn) throws Exception {
        try(Statement statement = conn.createStatement()) {
            if (ENTERPRISE) {
                statement.execute("CREATE TABLE IF NOT EXISTS `default`.`test` ON CLUSTER default (id Int64, name String) ENGINE = MergeTree() ORDER BY id;");
            } else {
                // Create a local table.
                statement.execute("CREATE TABLE IF NOT EXISTS `default`.`test_local` ON CLUSTER default (id Int64, name String) ENGINE = MergeTree() ORDER BY id;");
                // Create a distributed table.
                statement.execute("CREATE TABLE IF NOT EXISTS `default`.`test` ON CLUSTER default (id Int64, name String) ENGINE = Distributed(default, default, test_local, rand());");
            }
        }
    }

    /**
     * Insert data in batches.
     * @param conn The ClickHouse connection.
     * @param optimizeLevel The insert optimization level. 3 is faster than 2, and 2 is faster than 1.<br/>
     *                      1: insert into `default`.`test` (id, name) values(?, ?) -- with an additional query to get the table structure.
     *                         This is portable.<br/>
     *                      2: insert into `default`.`test` select id, name from input('id Int64, name String') -- effectively converts and inserts data sent to the server
     *                         with a given structure into the table with another structure. This is NOT portable because it is limited to ClickHouse.<br/>
     *                      3: insert into `default`.`test` format RowBinary -- fastest (close to the Java client) with streaming mode but requires manual serialization.
     *                         This is NOT portable because it is limited to ClickHouse.
     * @throws Exception
     */
    public static void batchInsert(Connection conn, int optimizeLevel) throws Exception {
        PreparedStatement preparedStatement = null;
        try {
            // Prepared statement
            switch (optimizeLevel) {
                case 1:
                    preparedStatement = conn.prepareStatement("insert into `default`.`test` (id, name) values(?, ?)");
                    break;
                case 2:
                    preparedStatement = conn.prepareStatement("insert into `default`.`test` select id, name from input('id Int64, name String')");
                    break;
                case 3:
                    preparedStatement = conn.prepareStatement("insert into `default`.`test` format RowBinary");
                    break;
                default:
                    throw new IllegalArgumentException("optimizeLevel must be 1, 2 or 3");
            }

            // Insert data.
            long randBase = (long) (Math.random() * 1000000); // A random number to prevent data duplication and loss.
            for (int i = 0; i < INSERT_BATCH_NUM; i++) {
                long insertStartTime = System.currentTimeMillis();
                switch (optimizeLevel) {
                    case 1:
                    case 2:
                        for (int j = 0; j < INSERT_BATCH_SIZE; j++) {
                            long id = (long) i * INSERT_BATCH_SIZE + j + randBase;
                            preparedStatement.setLong(1, id);
                            preparedStatement.setString(2, "name" + id);
                            preparedStatement.addBatch();
                        }
                        preparedStatement.executeBatch();
                        break;
                    case 3:
                        class MyClickHouseWriter implements ClickHouseWriter {
                            int batchIndex = 0;
                            public MyClickHouseWriter(int batchIndex) {
                                this.batchIndex = batchIndex;
                            }
                            @Override
                            public void write(ClickHouseOutputStream clickHouseOutputStream) throws IOException {
                                for (int j = 0; j < INSERT_BATCH_SIZE; j++) {
                                    long id = (long) batchIndex * INSERT_BATCH_SIZE + j + randBase;
                                    // Write id (Int64).
                                    ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
                                    buffer.order(ByteOrder.LITTLE_ENDIAN);
                                    buffer.putLong(id);
                                    clickHouseOutputStream.write(buffer.array());
                                    // Write name (String).
                                    clickHouseOutputStream.writeUnicodeString("name" + id);
                                }
                            }
                        }
                        preparedStatement.setObject(1, new MyClickHouseWriter(i));
                        preparedStatement.executeUpdate();
                        break;
                }

                System.out.printf("[%d] optimizeLevel=%d, insert batch [%d/%d] succeeded, cost %d ms\n",
                        Thread.currentThread().getId(), optimizeLevel, i + 1, INSERT_BATCH_NUM, System.currentTimeMillis() - insertStartTime);
            }
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }

    }

    /**
     * Count the table.
     * @param conn The ClickHouse connection.
     * @throws Exception
     */
    public static void count(Connection conn) throws Exception {
        try(Statement statement = conn.createStatement()) {
            ResultSet resultSet = statement.executeQuery("SELECT count() as cnt FROM `default`.`test`");
            if (resultSet.next()) {
                System.out.printf("Table `default`.`test` has %d rows\n", resultSet.getInt("cnt"));
            } else {
                throw new RuntimeException("Failed to count table `default`.`test`");
            }
        }
    }
}

mvn compile && mvn exec:java -Dexec.mainClass="com.aliyun.Main"

Table `default`.`test` has 500000 rows

properties.setProperty("socket_keepalive", "true");
properties.setProperty("http_connection_provider", "APACHE_HTTP_CLIENT");