HBase的etl初步完成

client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java

@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.client.adapter.hbase;
 
 import java.io.IOException;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -9,8 +10,11 @@ import javax.sql.DataSource;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.client.Connection;
-import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.alibaba.otter.canal.client.adapter.OuterAdapter;
 import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
@@ -29,12 +33,15 @@ import com.alibaba.otter.canal.client.adapter.support.*;
 @SPI("hbase")
 public class HbaseAdapter implements OuterAdapter {
 
-    private static volatile Map<String, MappingConfig> hbaseMapping       = null; // 文件名对应配置
-    private static volatile Map<String, MappingConfig> mappingConfigCache = null; // 库名-表名对应配置
+    private static Logger                              logger             = LoggerFactory.getLogger(HbaseAdapter.class);
+
+    private static volatile Map<String, MappingConfig> hbaseMapping       = null;                                       // 文件名对应配置
+    private static volatile Map<String, MappingConfig> mappingConfigCache = null;                                       // 库名-表名对应配置
 
     private Connection                                 conn;
     private HbaseSyncService                           hbaseSyncService;
     private HbaseTemplate                              hbaseTemplate;
+    private Configuration                              hbaseConfig;
 
     @Override
     public void init(OuterAdapterConfig configuration) {
@@ -55,7 +62,7 @@ public class HbaseAdapter implements OuterAdapter {
 
             Map<String, String> propertites = configuration.getProperties();
 
-            Configuration hbaseConfig = HBaseConfiguration.create();
+            hbaseConfig = HBaseConfiguration.create();
             propertites.forEach(hbaseConfig::set);
             conn = ConnectionFactory.createConnection(hbaseConfig);
             hbaseTemplate = new HbaseTemplate(conn);
@@ -90,6 +97,28 @@ public class HbaseAdapter implements OuterAdapter {
         }
     }
 
+    @Override
+    public Map<String, Object> count(String task) {
+        MappingConfig config = hbaseMapping.get(task);
+        String hbaseTable = config.getHbaseOrm().getHbaseTable();
+        long rowCount = 0L;
+        try {
+            HTable table = new HTable(hbaseConfig, hbaseTable);
+            Scan scan = new Scan();
+            scan.setFilter(new FirstKeyOnlyFilter());
+            ResultScanner resultScanner = table.getScanner(scan);
+            for (Result result : resultScanner) {
+                rowCount += result.size();
+            }
+        } catch (IOException e) {
+            logger.error(e.getMessage(), e);
+        }
+        Map<String, Object> res = new LinkedHashMap<>();
+        res.put("hbaseTable", hbaseTable);
+        res.put("count", rowCount);
+        return res;
+    }
+
     @Override
     public void destroy() {
         if (conn != null) {

client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java

@@ -72,6 +72,7 @@ public class MappingConfigLoader {
                 String[] dbTable;
                 if (dsKey == null) {
                     dbTable = srcMeta.split("\\.");
+
                 } else {
                     dbTable = srcMeta.split("@")[0].split("\\.");
                 }
@@ -84,7 +85,7 @@ public class MappingConfigLoader {
                     hbaseOrm.setAutoCreateTable(true);
                     hbaseOrm.setDatabase(dbTable[0]);
                     hbaseOrm.setTable(dbTable[1]);
-                    hbaseOrm.setMode(MappingConfig.Mode.STRING);
+                    hbaseOrm.setMode(MappingConfig.Mode.PHOENIX);
                     hbaseOrm.setRowKey(rowKey);
                     // 有定义rowKey
                     if (rowKey != null) {
@@ -94,6 +95,7 @@ public class MappingConfigLoader {
                         hbaseOrm.setRowKeyColumn(columnItem);
                     }
                     config.setHbaseOrm(hbaseOrm);
+                    config.setDataSourceKey(dsKey);
 
                 } else {
                     throw new RuntimeException(String.format("配置项[%s]内容为空, 或格式不符合database.table", c));

client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java

@@ -68,7 +68,7 @@ public class HbaseEtlService {
             // 判断hbase表是否存在，不存在则建表
             MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
             if (!hbaseTemplate.tableExists(hbaseOrm.getHbaseTable())) {
-                hbaseTemplate.createTable(hbaseOrm.getHbaseTable(), hbaseOrm.getFamilies().toArray(new String[0]));
+                hbaseTemplate.createTable(hbaseOrm.getHbaseTable(), hbaseOrm.getFamily());
             }
         } catch (Exception e) {
             logger.error(e.getMessage(), e);

client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml

@@ -3,7 +3,7 @@ hbaseOrm:
   mode: PHOENIX  #NATIVE   #STRING
   database: mytest  # 数据库名
   table: person2     # 数据库表名
-  hbaseTable: MYTEST_PERSON2   # HBase表名
+  hbaseTable: MYTEST.PERSON2   # HBase表名
   family: CF  # 默认统一Family名称
   uppercaseQualifier: true  # 字段名转大写, 默认为true
   commitBatch: 3000 # 批量提交的大小