ApsaraDB for ClickHouse:Connect using JDBC

<dependency>
    <groupId>com.zaxxer</groupId>
    <artifactId>HikariCP</artifactId>
    <version>3.4.5</version>
</dependency>
<dependency>
    <groupId>com.clickhouse</groupId>
    <artifactId>clickhouse-jdbc</artifactId>
    <version>0.4.6</version>
</dependency>
<dependency>
    <groupId>org.lz4</groupId>
    <artifactId>lz4-java</artifactId>
    <version>1.8.0</version>
</dependency>
<dependency>
    <groupId>org.apache.httpcomponents.client5</groupId>
    <artifactId>httpclient5</artifactId>
    <version>5.2.1</version>
</dependency>

Complete sample code

The following sample code shows how to create a table named test in the default database of an Enterprise Edition cluster and concurrently insert 10 batches of data, with 10,000 rows per batch.

Before you run the code, modify the parameters as needed for your business scenario. For more information about the parameters, see the parameter table in the Code flow section.

The main logic and entry point of this code is the main method.

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`");
            }
        }
    }
}