劍道第一仙

公告

chatgpt使用記錄

網址：http://chat.yangxiwl.cn/chatbotai/1002?channel=bdtoufangChatztHD3

寫一個java程序，要求將kafka的數據消費批量插入的clickhouse數據庫

你可以使用Kafka的Java客戶端庫和ClickHouse的Java驅動來實現將Kafka數據消費并批量插入到ClickHouse數據庫中的功能。以下是一個簡單的示例程序：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import ru.yandex.clickhouse.BalancedClickhouseDataSource;
import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHousePreparedStatement;
import ru.yandex.clickhouse.settings.ClickHouseProperties;

import java.sql.SQLException;
import java.util.*;

public class KafkaToClickHouse {

    public static void main(String[] args) {
        // Kafka配置
        Properties kafkaProps = new Properties();
        kafkaProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        kafkaProps.put(ConsumerConfig.GROUP_ID_CONFIG, "clickhouse-consumer");
        kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        kafkaProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // ClickHouse配置
        ClickHouseProperties clickHouseProps = new ClickHouseProperties();
        clickHouseProps.setHost("localhost");
        clickHouseProps.setPort(8123);
        clickHouseProps.setDatabase("your_database_name");
        clickHouseProps.setUser("your_username");
        clickHouseProps.setPassword("your_password");

        // Kafka消費者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafkaProps);
        consumer.subscribe(Collections.singletonList("your_topic_name"));

        // ClickHouse連接池
        BalancedClickhouseDataSource dataSource = new BalancedClickhouseDataSource("jdbc:clickhouse://", clickHouseProps);

        // 消費并插入數據
        try (ClickHouseConnection connection = dataSource.getConnection();
             ClickHousePreparedStatement stmt = connection.prepareStatement("INSERT INTO your_table_name (col1, col2) VALUES (?, ?)")) {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    String[] data = record.value().split(","); // 假設數據格式為逗號分隔的兩列
                    stmt.setString(1, data[0]);
                    stmt.setString(2, data[1]);
                    stmt.addBatch();
                }
                stmt.executeBatch();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            consumer.close();
            dataSource.close();
        }
    }
}

下面是一個示例程序，它將Kafka的數據按照每100條記錄進行批量插入到ClickHouse數據庫，并在每個批次結束后提交事務

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import ru.yandex.clickhouse.BalancedClickhouseDataSource;
import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHousePreparedStatement;
import ru.yandex.clickhouse.settings.ClickHouseProperties;

import java.sql.SQLException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class KafkaToClickHouse {

    public static void main(String[] args) {
        // Kafka配置
        Properties kafkaProps = new Properties();
        kafkaProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        kafkaProps.put(ConsumerConfig.GROUP_ID_CONFIG, "clickhouse-consumer");
        kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        kafkaProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100); // 每次最多拉取100條記錄

        // ClickHouse配置
        ClickHouseProperties clickHouseProps = new ClickHouseProperties();
        clickHouseProps.setHost("localhost");
        clickHouseProps.setPort(8123);
        clickHouseProps.setDatabase("your_database_name");
        clickHouseProps.setUser("your_username");
        clickHouseProps.setPassword("your_password");

        // Kafka消費者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafkaProps);
        consumer.subscribe(Collections.singletonList("your_topic_name"));

        // ClickHouse連接池
        BalancedClickhouseDataSource dataSource = new BalancedClickhouseDataSource("jdbc:clickhouse://", clickHouseProps);

        // 消費并插入數據
        try (ClickHouseConnection connection = dataSource.getConnection()) {
            List<String> batch = new ArrayList<>();
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    String[] data = record.value().split(","); // 假設數據格式為逗號分隔的兩列
                    String row = String.format("('%s', '%s')", data[0], data[1]);
                    batch.add(row);
                }
                if (batch.size() >= 100) {
                    insertBatch(connection, batch);
                    batch.clear();
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            consumer.close();
            dataSource.close();
        }
    }

    private static void insertBatch(ClickHouseConnection connection, List<String> batch) throws SQLException {
        try (ClickHousePreparedStatement stmt = connection.prepareStatement("INSERT INTO your_table_name (col1, col2) VALUES " + String.join(",", batch))) {
            stmt.execute();
        }
    }
}

如何實現批量提交的同時避免重復插入

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import ru.yandex.clickhouse.BalancedClickhouseDataSource;
import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHousePreparedStatement;
import ru.yandex.clickhouse.settings.ClickHouseProperties;

import java.sql.SQLException;
import java.time.Duration;
import java.util.*;

public class KafkaToClickHouse {

    public static void main(String[] args) {
        // Kafka配置
        Properties kafkaProps = new Properties();
        kafkaProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        kafkaProps.put(ConsumerConfig.GROUP_ID_CONFIG, "clickhouse-consumer");
        kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        kafkaProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100); // 每次最多拉取100條記錄

        // ClickHouse配置
        ClickHouseProperties clickHouseProps = new ClickHouseProperties();
        clickHouseProps.setHost("localhost");
        clickHouseProps.setPort(8123);
        clickHouseProps.setDatabase("your_database_name");
        clickHouseProps.setUser("your_username");
        clickHouseProps.setPassword("your_password");

        // Kafka消費者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafkaProps);
        consumer.subscribe(Collections.singletonList("your_topic_name"));

        // ClickHouse連接池
        BalancedClickhouseDataSource dataSource = new BalancedClickhouseDataSource("jdbc:clickhouse://", clickHouseProps);

        // 消費并插入數據
        try (ClickHouseConnection connection = dataSource.getConnection()) {
            List<String> batch = new ArrayList<>();
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    String[] data = record.value().split(","); // 假設數據格式為逗號分隔的兩列
                    String row = String.format("('%s', '%s')", data[0], data[1]);
                    batch.add(row);
                }
                if (batch.size() >= 100) {
                    insertBatch(connection, batch);
                    batch.clear();
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            consumer.close();
            dataSource.close();
        }
    }

    private static void insertBatch(ClickHouseConnection connection, List<String> batch) throws SQLException {
        try (ClickHousePreparedStatement stmt = connection.prepareStatement("INSERT INTO your_table_name (col1, col2) VALUES " + String.join(",", batch))) {
            stmt.execute();
        } catch (SQLException e) {
            // 處理重復插入的異常
            if (e.getMessage().contains("Code: 409")) {
                // 重復插入數據，可以忽略或者進行其他處理
            } else {
                throw e;
            }
        }
    }
}

posted on 2023-10-10 22:45 劍道第一仙閱讀(34) 評論(0) 收藏舉報

刷新頁面返回頂部

劍道第一仙

導航

公告

chatgpt使用記錄