ApsaraDB for ClickHouse：Menghubungkan menggunakan 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>

Kode contoh lengkap

Kode contoh berikut menunjukkan cara membuat tabel bernama test di database default kluster Edisi Perusahaan dan memasukkan data secara konkuren sebanyak 10 batch, dengan 10.000 baris per batch.

Sebelum menjalankan kode, ubah parameter sesuai kebutuhan skenario bisnis Anda. Untuk informasi selengkapnya tentang parameter, lihat tabel parameter di bagian Alur kode.

Logika utama dan titik masuk kode ini 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.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"; // KONFIGURASI ANDA DI SINI
    private final static String DATABASE = "default"; // 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/%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 {
            HikariConfig conf = buildHikariDataSource();
            try(HikariDataSource ds = new HikariDataSource(conf)) {
                // Buat tabel.
                Connection conn = ds.getConnection();
                createTable(conn);
                conn.close();

                // Masukkan data secara konkuren.
                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 mulai memasukkan\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 berhenti memasukkan\n", Thread.currentThread().getId());
                            countDownLatch.countDown();
                        }
                    });
                }
                // Tunggu hingga semua thread selesai.
                countDownLatch.await();

                // Hitung jumlah data di tabel.
                conn = ds.getConnection();
                count(conn);
                conn.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * Hasilkan URL JDBC.
     * @param protocol Protokol. Protokol yang didukung termasuk http, https, dan grpc.
     * @param endpoint Titik akhir.
     * @return URL JDBC.
     */
    public static String getJdbcUrl(String protocol, String endpoint, String database) {
        return String.format(JDBC_URL, protocol, endpoint, database);
    }

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

        // Properti
        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");
        /// Zona waktu
        properties.setProperty("use_server_time_zone", "true");

        // Konfigurasi sumber data
        conf.setDataSource(new ClickHouseDataSource(getJdbcUrl(YOUR_INSTANCE_PROTOCOL, YOUR_INSTANCE_ENDPOINT, DATABASE), properties));
        conf.setUsername(YOUR_INSTANCE_USER);
        conf.setPassword(YOUR_INSTANCE_PASSWORD);

        // Konfigurasi kolam koneksi
        conf.setMaximumPoolSize(10);
        conf.setMinimumIdle(5);
        conf.setIdleTimeout(30000);
        conf.setMaxLifetime(60000);
        conf.setConnectionTimeout(30000);
        conf.setPoolName("HikariPool");

        return conf;
    }

    /**
     * Buat tabel.
     * @param conn Koneksi ClickHouse.
     * @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 {
                // Buat tabel lokal.
                statement.execute("CREATE TABLE IF NOT EXISTS `default`.`test_local` ON CLUSTER default (id Int64, name String) ENGINE = MergeTree() ORDER BY id;");
                // Buat tabel terdistribusi.
                statement.execute("CREATE TABLE IF NOT EXISTS `default`.`test` ON CLUSTER default (id Int64, name String) ENGINE = Distributed(default, default, test_local, rand());");
            }
        }
    }

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

            // Masukkan data.
            long randBase = (long) (Math.random() * 1000000); // Bilangan acak untuk mencegah duplikasi dan kehilangan 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 name (String).
                                    clickHouseOutputStream.writeUnicodeString("name" + id);
                                }
                            }
                        }
                        preparedStatement.setObject(1, new MyClickHouseWriter(i));
                        preparedStatement.executeUpdate();
                        break;
                }

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

    }

    /**
     * Hitung jumlah data di tabel.
     * @param conn Koneksi ClickHouse.
     * @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("Tabel `default`.`test` memiliki %d baris\n", resultSet.getInt("cnt"));
            } else {
                throw new RuntimeException("Gagal menghitung tabel `default`.`test`");
            }
        }
    }
}