kafka topic 发送消息到 hbase_虚拟机kafka到hbase-CSDN博客

本文链接：https://blog.csdn.net/Xiayebuliang/article/details/130488286

该代码示例展示了如何配置Kafka消费者从指定主题消费数据，然后将数据处理后存入HBase数据库。首先配置Kafka消费者连接信息，然后设置HBase的连接参数。数据处理过程中，将消费到的用户ID和朋友ID拆分并创建Put对象写入HBase表。程序在一个无限循环中持续消费和处理数据，每次处理一批数据后更新计数并休眠10毫秒。

摘要生成于 C知道，由 DeepSeek-R1 满血版支持，前往体验 >

一、配置kafka信息

二、配置hbase信息，连接hbase数据库

需要先在hbase创建对应的命令空间和table

Put对象，按照不同需求，从文件中截取对应字段

三、完整代码

一、配置kafka信息

        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.91.180:9092");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");  // 手动提交
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");   // 自动提交时，提交时间
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "userfriend_group2");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 配置Kafka信息。消费消息
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singleton("user_friends"));

二、配置hbase信息，连接hbase数据库

需要先在hbase创建对应的命令空间和table

Put对象，按照不同需求，从文件中截取对应字段

     // 配置Kafka信息。消费消息
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);