MongoDB數據導入hbase + 代碼

需求：
從mongoDB里面查出來數據，判斷是否有該列簇，如果有則導入此條數據+列簇，如果沒有，則該條數據不包含該列簇

直接貼出代碼：

package Test;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

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;
import org.bson.Document;

import com.mongodb.MongoClient;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;

public class OperateTable2 {
private static Configuration conf = null;
static {
conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum","master.hadoop");
conf.set("hbase.zookeeper.property.clientPort", "2181");
conf.set("hbase.master", "master.hadoop:60000");
}
public static void addRow(String tableName, String row,
            String columnFamily, String column, String value) throws Exception {
        HTable hTable = new HTable(conf, tableName);
        Put put = new Put(Bytes.toBytes(row));
        // 參數出分別：列族、列、值
        put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column),
                Bytes.toBytes(value));
        hTable.put(put);
 
    }
public static void main(String[] args) {
// 聲明靜態配置
String tableName = "house";
        String columnFamilys = "info";
        int a = 0;
try {
ServerAddress serverAddress = new ServerAddress("42.62.66.9",27017);
List addrs = new ArrayList();
addrs.add(serverAddress);
MongoCredential credential = MongoCredential.createScramSha1Credential("admin", 
"admin", "bigmaster654321".toCharArray());
List credentials = new ArrayList();
credentials.add(credential);
//通過認證獲取mongoDB的連接
MongoClient mongoClient = new MongoClient(addrs,credentials);
MongoDatabase mongoDatabase = mongoClient.getDatabase("house");
System.out.println("MongoDB Connection-----------------Successfully");
MongoCollection collection = mongoDatabase.getCollection("houses2");
FindIterable findIterable  = collection.find();
MongoCursor mongoCursor = findIterable.iterator(); 
while(mongoCursor.hasNext()){
a=a+1;
Document document = mongoCursor.next();
   Set set = document.keySet();
Iterator it = set.iterator();
while(it.hasNext()){
 String tags = it.next();
 if(tags.equals("_id")){
 continue;
 }
 OperateTable2.addRow(tableName, document.getString("_id"), columnFamilys, tags, document.getString(tags));
}
System.out.println("Insert Into HBase Success"+"This is the "+a+" data");
}
System.out.println("Compelete All Insert");
} catch (Exception e) {
// TODO: handle exception
System.err.println(e.getClass().getName()+ ":" +e.getMessage());
}
}
}

我這里是用_id為hbase的rowkey，列為info，代碼無誤，根據需求可以改動，關鍵在於兩個while判斷的地方，那里最容易出錯,導入mysql也可以用此代碼改動取值的地方和jdbc連接進行導入

需要用到的包：mongo-java-driver-3.2.2.jar，hbase常用的包可以全導

代碼如上，原創轉載注明出處！