我們在使用kafka處理資料的過程中會使用kafka跟一下資料庫進行互動，Hbase就是其中的一種。下面給大家介紹一下kafka中的資料是如何匯入Hbase的。

本文的思路是通過consumers把資料消費到Hbase中。

首先在Hbase中建立表，建立表可以在Hbase客戶端建立也可以通過API建立，這裡介紹通過API建立表的方法：

建立CreatTableTest類

import java.io.IOException; 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.hbase.HBaseConfiguration; 
import org.apache.hadoop.hbase.HColumnDescriptor; 
import org.apache.hadoop.hbase.HTableDescriptor; 
import org.apache.hadoop.hbase.client.HBaseAdmin;
public class CreatTableTest {
	public static void main(String[] args) throws IOException  { 
		//設定HBase據庫的連線配置引數
		Configuration conf = HBaseConfiguration.create(); 
		conf.set("hbase.zookeeper.quorum",  "192.168.5.128");  //  Zookeeper的地址
		conf.set("hbase.zookeeper.property.clientPort", "42182"); 
		String tableName = "emp";
		String[] family = { "basicinfo","deptinfo"}; 
		HBaseAdmin hbaseAdmin = new HBaseAdmin(conf); 
		//建立表物件
		HTableDescriptor hbaseTableDesc = new HTableDescriptor(tableName); 
		for(int i = 0; i < family.length; i++) { 
		//設定表字段
		hbaseTableDesc.addFamily(new HColumnDescriptor(family[i])); 

		} 
		//判斷表是否存在，不存在則建立，存在則列印提示資訊
		if(hbaseAdmin.tableExists(tableName)) { 
		System.out.println("TableExists!"); 
		System.exit(0); 
		} else{ 
		hbaseAdmin.createTable(hbaseTableDesc); 
		System.out.println("Create table Success!"); 
		} 
	} 
}
 

建立表之後我們建立一個consumer來消費資料到Hbase中

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import com.teamsun.kafka.m001.KafkaProperties;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class KafkaConsumer3 extends Thread {
	private final ConsumerConnector consumer;
	private final String topic;

	public KafkaConsumer3(String topic) {
		consumer = kafka.consumer.Consumer
				.createJavaConsumerConnector(createConsumerConfig());
		this.topic = topic;
	}

	private static ConsumerConfig createConsumerConfig() {
		Properties props = new Properties();
		props.put("zookeeper.connect", KafkaProperties.zkConnect);
		props.put("group.id", KafkaProperties.groupId1);
		props.put("zookeeper.session.timeout.ms", "40000");
		props.put("zookeeper.sync.time.ms", "200");
		props.put("auto.commit.interval.ms", "1000");
		return new ConsumerConfig(props);
	}

	@Override
	public void run() {
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, new Integer(1));
		Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer
				.createMessageStreams(topicCountMap);
		KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
		ConsumerIterator<byte[], byte[]> it = stream.iterator();
	    HBaseUtils hbase = new HBaseUtils();  
		while (it.hasNext()) {
			System.out.println("3receive：" + new String(it.next().message()));
			try {
				hbase.put(new String(it.next().message()));
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
//			try {
//				sleep(300);    // 每條訊息延遲300ms
//			} catch (InterruptedException e) {
//				e.printStackTrace();
//			}
		}
	}
}
 

再建立一個HBaseUtils來指定要連線的Hbase資料庫

import java.io.IOException;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
public class HBaseUtils {
	public  void put(String string) throws IOException { 
	//設定HBase據庫的連線配置引數
	Configuration conf = HBaseConfiguration.create(); 
	conf.set("hbase.zookeeper.quorum",  "192.168.5.128");  //  Zookeeper的地址
	conf.set("hbase.zookeeper.property.clientPort", "42182"); 
	Random random = new Random();
	long a = random.nextInt(1000000000);           
	String tableName = "emp"; 
	String rowkey = "rowkey"+a ;
	String columnFamily = "basicinfo"; 
	String column = "empname"; 
	//String value = string; 
	HTable table=new HTable(conf, tableName); 
	Put put=new Put(Bytes.toBytes(rowkey)); 
	put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(string)); 
	table.put(put);//放入表
	table.close();//釋放資源
	}
}
 

最後再加上consumer的配置檔案就大功告成了。

public interface KafkaProperties {

	final static String zkConnect = "hadoop0:42182,hadoop1:42182,hadoop2:42182,hadoop3:42182";
	final static String groupId1= "group1";
	final static String topic = "test3";
	final static String kafkaServerURL = "hadoop0,hadoop1";
	final static int kafkaServerPort = 9092;
	final static int kafkaProducerBufferSize = 64 * 1024;
	final static int connectionTimeOut = 20000;
	final static int reconnectInterval = 10000;
	final static String clientId = "SimpleConsumerDemoClient";
}

然後執行consumer就可以了，注意要保證topic中有訊息才可以消費。

public class KafkaConsumerProducerTest {

	public static void main(String[] args) {
//		 KafkaProducer1 producerThread1 = new KafkaProducer1(KafkaProperties.topic);
//       producerThread1.start();
//       KafkaProducer2 producerThread2 = new KafkaProducer2(KafkaProperties.topic);
//       producerThread2.start();
//       KafkaProducer3 producerThread3 = new KafkaProducer3(KafkaProperties.topic);
//       producerThread3.start();
        
//       KafkaConsumer1 consumerThread1 = new KafkaConsumer1(KafkaProperties.topic);
//       consumerThread1.start();
//       KafkaConsumer2 consumerThread2 = new KafkaConsumer2(KafkaProperties.topic);
//       consumerThread2.start();
         KafkaConsumer3 consumerThread3 = new KafkaConsumer3(KafkaProperties.topic);
         consumerThread3.start();
//       KafkaConsumer4 consumerThread4 = new KafkaConsumer4(KafkaProperties.topic);
//       consumerThread4.start();
	}
}

在HBase客戶端執行

            hbase(main):063:0> scan  'emp'  

就可以檢視到資料了。

以上就是kafka資料進入Hbase的一個例子，當然上訴只是保證資料走通了，大家在具體專案中什麼需求，還需要自行修改和完善。

——————————————————————————————————

作者：桃花惜春風

轉載請標明出處，原文地址：  

如果感覺本文對您有幫助，請留下您的贊，您的支援是我堅持寫作最大的動力，謝謝！