kafkaの膨大なデータをファイルに書き込むにはどうすれば

2年 ago
夏樹, 風
2 minutes

Kafkaから大量のデータをファイルに書き出すには、KafkaのConsumer APIを使用してデータを消費し、ファイルを書き出すことができます。

以下の Java で記述されたサンプルのプログラムは、Kafka からデータを消費して、ファイルに書き込むものです。

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 java.io.FileWriter;
import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaToFile {
    public static void main(String[] args) {
        // Kafka配置
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 创建Kafka消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList("test-topic"));

        try {
            // 创建文件写入器
            FileWriter writer = new FileWriter("output.txt");

            while (true) {
                // 拉取数据
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    // 将数据写入文件
                    writer.write(record.value());
                    writer.write("\n");
                }

                // 刷新缓冲区
                writer.flush();
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 关闭文件写入器和消费者
            try {
                writer.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            consumer.close();
        }
    }
}

上記のサンプルコードでは、まずKafkaの設定に基づいてKafkaコンシューマーを作成します。 次に、コンシューマーするテーマ（例えば、「test-topic」）をサブスクライブします。 続いて、データをファイルに書き込むためのファイルライターを作成します。 その後、無限ループに入ります。 各ループで、Kafkaからpoll()メソッドを介してデータを取得し、データをファイルに書き込みます。 最後に、プログラム終了時にファイルライターとコンシューマーを終了します。

このプログラムを実行するには、プロジェクトに Kafka の依存関係を追加する必要があります。Maven プロジェクトでは、次の依存関係を追加できます。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
</dependency>

上記のサンプルプログラムをJavaファイルとして保存し、適切なビルドツール（Mavenなど）を使用してビルドして実行します。プログラムが実行されると、Kafkaからデータを取得し、「output.txt」というファイルに書き込みます。

bannerAds