Linux平台HBase与Kafka集成（kafka到hbase）

<配置>
 <属性>
 <名称>bootstrap.servers</名称>
 <值>localhost:9092</值>
  </属性>
 <属性rty>
 <名称>group.id</名称>
 <值>hbase_consumer_group</值>
 </属性>
 <属性>
 <名称>键.deserializer</名称>
 <值>org.apache.kafka.common.sSerialization.StringDeserializer</值>
 </ 属性>
 <属性> 
 <名称>value.deserializer</名称>
 <值>org.apache.kafka.common.serialization.StringDeserializer</值> 
 </属性>
 <合适ty>
 <名称>auto.offset.reset</名称>
 <值>最旧的</值>
 </属性>
 <属性>
 <名称>enable.auto.commit</名称>
 <值>false</值>
 </属性>
 <属性>
 <名称>auto.commit.interval.ms</名称>
 <值>1000 </值>
 </属性>
</配置>

4.测试集成

完成上述配置后，就可以创建一个简单的测试程序了。集成工作正常。 下面是一个 Java 程序示例：

import org.apache.hadoop.conf.Configuration ； 
导入 org.apache.hadoop.hbase.HBaseConfiguration;
导入 org.apache. hadoop.hbase.KeyValue;
导入 org.apache.hadoop.hbase.TableName;
导入跨度>跨度>跨度> org.apache.hadoop.hbase.client.*;
导入 org.apache.kafka.clients.Producer.KafkaProducer;
导入 org.apache.kafka.clients.Producer.ProducerRecord;
导入 org.apache.卡夫卡。  客户。    Consumer.ConsumerRecord;
导入 org.apache.kafka.clients.consumer.ConsumerRecords;
导入 org.apache.kafka.clients.consumer.KafkaConsumer;

导入 java.util.Arrays;
导入 java.util.Properties;

公共 类 HBaseKafkaIntegrationTest {
 public 静态 void main</span span>(String[] args) 抛出 异常 {
 配置 conf = HBaseConfiguration.create();
 连接 连接 = ConnectionFactory.createConnection(conf);
 管理员 管理员 = connection.getAdmin();

  // 创建表
 表名 表名 = TableName.valueOf("test_table");
 if  (!admin.tableExists(tableName)) {
 HTableDescriptor tableDescriptor = 新 HTableDescriptor (tableName);
 HColumnDescriptor columnDescriptor = 新 HColumnDescriptor("cf1 ");
 tableDescriptor.addFamily(columnDescriptor);
 admin.createTable(tableDescriptor);
 }

 // 插入数据到HBase
 表 表  = 连接.getTable(tableName);
 放置 放置 = 新 地点("row1".getBytes());
 put.addColumn("cf1".getBytes(), "column1".getBytes(), "value1".getBytes());
 table.put(put);

 // Kafka将数据发送到
 Properties ProducerProps = 新建 属性();
ProducerProps.put("bootstrap.servers", "localhost ) :9092");
 Producer Props.put("key.serializer", "org .apache .kafka.common.serialization.StringSerializer");
ProducerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 KafkaProducer Producer = new KafkaProducer(ProducerProps);
 生产者发送(new ProducerRecord("hbase_kafka_topic", "row1", "value1"));
 Producer.close();

 //从Kafka消费数据
 Properties consumerProps  = 新 属性();
 ConsumerProps.put("bootstrap.servers", "localhost ) :9092");
 ConsumerProps.put("group.id", "hbase_consumer_group");
 ConsumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 ConsumerProps.put("value.deserializer" , "org.apache.kafka.common.serialization.StringDeserializer");
 KafkaConsumer Consumer = new KafkaConsumer(consumerProps);
 Consumer.subscribe(Arrays.asList(") hbase_kafka_topic"));

 之间 (true) {
 ConsumerRecords record = Consumer.poll(100);
 for (ConsumerRecord record : record) {
 System.out.printf("offset = % d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); 
 // 处理记录并将其放入HBase
 Put put = 新建 放入(record.key () .getBytes());
 put.addColumn("cf1".getBytes(), "column1".getBytes(), record.value().getBytes()) ;
 table.put(put);
 }
 }
 }
}

手动运行上面的程序。 , 验证HBase和Kafka服务是否正常运行，并观察输出日志，确保数据从Kafka正确传输到HBase。

总结

通过以上步骤，您可以在Linux平台上成功集成HBase和Kafka。 这种集成方式可以实现实时数据处理和数据流存储，适合需要高性能、高吞吐量的应用场景。