ApsaraDB for ClickHouse：Gunakan JDBC untuk menghubungkan ke cluster ApsaraDB for ClickHouse

<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>

Kode contoh lengkap

Dalam database default dari kluster Enterprise Edition, buat tabel bernama test. Kemudian, sisipkan data ke dalam tabel test secara bersamaan dalam batch. Setiap batch berisi 10.000 baris. Lakukan operasi ini total sebanyak 10 batch.

Untuk menjalankan kode, Anda harus memodifikasi parameter terkait berdasarkan kebutuhan bisnis Anda. Untuk informasi lebih lanjut tentang parameter, lihat bagian Proses Keseluruhan dari topik ini.

Logika utama dan titik masuk kode adalah metode main.

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.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"; // KONFIGURASI ANDA DI SINI
    private final static String YOUR_INSTANCE_USER = "USER"; // KONFIGURASI ANDA DI SINI
    private final static String YOUR_INSTANCE_PASSWORD = "PASSWORD"; // KONFIGURASI ANDA DI SINI
    private final static String JDBC_URL = "jdbc:clickhouse:%s://%s";
    private final static Integer INSERT_BATCH_SIZE = 10000;
    private final static Integer INSERT_BATCH_NUM = 10;
    private final static boolean ENTERPRISE = true; // KONFIGURASI ANDA DI SINI
    private final static Integer INSERT_OPTIMIZE_LEVEL = 3;

    public static void main(String[] args) {
        try {
            // 1. Hubungkan ke database. 
            HikariConfig conf = buildHikariDataSource();
            try(HikariDataSource ds = new HikariDataSource(conf)) {
                Connection conn = ds.getConnection();

                // 2. Buat tabel. 
                createTable(conn);

                // 3. Sisipkan data secara bersamaan. 
                int concurrentNum = 5;
                // Mulai lima thread. 
                CountDownLatch countDownLatch = new CountDownLatch(concurrentNum);
                ExecutorService executorService = Executors.newFixedThreadPool(concurrentNum);
                for (int i = 0; i < concurrentNum; i++) {
                    executorService.submit(() -> {
                        System.out.printf("[%d] Thread mulai menyisipkan\n", Thread.currentThread().getId());
                        try {
                            //Sisipkan data. 
                            batchInsert(ds.getConnection(), INSERT_OPTIMIZE_LEVEL);
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            System.out.printf("[%d] Thread berhenti menyisipkan\n", Thread.currentThread().getId());
                            countDownLatch.countDown();
                        }
                    });
                }
                // Tunggu hingga setiap thread menyelesaikan penyisipan data. 
                countDownLatch.await();

                // 4. Lihat hasilnya. 
                count(conn);
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * generate JDBC URL
     * @param protocol mendukung http, https, grpc
     * @param endpoint endpoint
     * @return JDBC URL
     */
    public static String getJdbcUrl(String protocol, String endpoint) {
        return String.format(JDBC_URL, protocol, endpoint);
    }

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

        // datasource config
        conf.setDataSource(new ClickHouseDataSource(getJdbcUrl(YOUR_INSTANCE_PROTOCOL, YOUR_INSTANCE_ENDPOINT)));
        conf.setUsername(YOUR_INSTANCE_USER);
        conf.setPassword(YOUR_INSTANCE_PASSWORD);

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

        return conf;
    }

    /**
     * create table
     * @param conn Koneksi ClickHouse
     * @throws Exception
     */
    public static void createTable(Connection conn) throws Exception {
        if (ENTERPRISE) {
            conn.createStatement().execute("CREATE TABLE IF NOT EXISTS `default`.`test` ON CLUSTER default (id Int64, name String) ENGINE = MergeTree() ORDER BY id;");
        } else {
            // buat tabel lokal
            conn.createStatement().execute("CREATE TABLE IF NOT EXISTS `default`.`test_local` ON CLUSTER default (id Int64, name String) ENGINE = MergeTree() ORDER BY id;");
            // buat tabel terdistribusi
            conn.createStatement().execute("CREATE TABLE IF NOT EXISTS `default`.`test` ON CLUSTER default (id Int64, name String) ENGINE = Distributed(default, default, test_local, rand());");
        }
    }

    /**
 * batch insert
 * @param conn Koneksi ClickHouse
 * @param optimizeLevel tingkat optimasi penyisipan, 3 lebih cepat dari 2, 2 lebih cepat dari 1<br/>
 *                      1: insert into `default`.`test` (id, name) values(?, ?) -- dengan query tambahan untuk mendapatkan struktur tabel.
 *                         Ini portabel.<br/>
 *                      2: insert into `default`.`test` select id, name from input('id Int64, name String') -- efektif mengubah dan menyisipkan data yang dikirim ke server
 *                         dengan struktur tertentu ke tabel dengan struktur lain. Ini TIDAK portabel (terbatas pada ClickHouse).<br/>
 *                      3: insert into `default`.`test` format RowBinary -- paling cepat (mendekati Java client) dengan mode streaming tetapi memerlukan serialisasi manual dan ini
 *                         TIDAK portabel (terbatas pada ClickHouse).
 * @throws Exception
 */
    public static void batchInsert(Connection conn, int optimizeLevel) throws Exception {
        // prepared statement
        PreparedStatement preparedStatement = null;
        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 harus 1, 2 atau 3");
        }

        // sisipkan data
        long randBase = (long) (Math.random() * 1000000); // nomor acak, mencegah duplikasi dan hilangnya data
        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;
                                // tulis id(Int64)
                                ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
                                buffer.order(ByteOrder.LITTLE_ENDIAN);
                                buffer.putLong(id);
                                clickHouseOutputStream.write(buffer.array());
                                // tulis nama(String)
                                clickHouseOutputStream.writeUnicodeString("name" + id);
                            }
                        }
                    }
                    preparedStatement.setObject(1, new MyClickHouseWriter(i));
                    preparedStatement.executeUpdate();
                    break;
            }

            System.out.printf("[%d] optimizeLevel=%d, sisipkan batch [%d/%d] berhasil, biaya %d ms\n",
                    Thread.currentThread().getId(), optimizeLevel, i + 1, INSERT_BATCH_NUM, System.currentTimeMillis() - insertStartTime);
        }
    }

    /**
     * hitung tabel
     * @param conn Koneksi ClickHouse
     * @throws Exception
     */
    public static void count(Connection conn) throws Exception {
        ResultSet resultSet = conn.createStatement().executeQuery("SELECT count() as cnt FROM `default`.`test`");
        if (resultSet.next()) {
            System.out.printf("tabel `default`.`test` memiliki %d baris\n", resultSet.getInt("cnt"));
        } else {
            throw new RuntimeException("gagal menghitung tabel `default`.`test`");
        }
    }
}