Kafka---将kafka中的数据导入HBase

Posted 2021-06-15 Shall潇

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
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 java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

/**
 * @Author shall潇
 * @Date 2021/5/31
 * @Description  将 kafka topic数据导入 hbase
 */
public class UserFriendToHB {
    static int num = 0;
    public static void main(String[] args) {
        //1.kafka 消费端属性配置
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.159.100:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,"30000");

        //是否自动提交，获取数据的状态 false 是手动提交，true是自动提交，时间间隔1000
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"user_friend_group");
        
        //订阅对应主题的消息
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singleton("user_friends"));

        //2.HBase 配置
        Configuration conf = HBaseConfiguration.create();
        conf.set(HConstants.HBASE_DIR,"hdfs://192.168.159.100:9000/hbase");
        conf.set("hbase.zookeeper.quorum","192.168.159.100");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        Connection connection = null;
        try {
            connection = ConnectionFactory.createConnection(conf);
            Admin admin = connection.getAdmin();
            Table table = connection.getTable(TableName.valueOf("event_db:user_friend"));

            while (true){
            	//3.处理读到的record,并且put到HBase中的表中
                ConsumerRecords<String, String> records = consumer.poll(100);
                List<Put> datas = new ArrayList<>();
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(record.value());
                    String[] split = record.value().split(",");
                    Put put = new Put(Bytes.toBytes((split[0] + split[1]).hashCode()));
                    put.addColumn("uf".getBytes(),"userid".getBytes(),split[0].getBytes());
                    put.addColumn("uf".getBytes(),"friendid".getBytes(),split[1].getBytes());
                    datas.add(put);
                }
                num+=datas.size();
                System.out.println(num+"行");
                table.put(datas);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}