Merge branch 'fix_hbase_sink_bug' into 'v1.8.0_dev'

回滚脏数据代码后，修复hbasesink bug



See merge request !159

Author: toutian

docs/hbaseSink.md

@@ -7,7 +7,7 @@ CREATE TABLE MyResult(
     type ='hbase',
     zookeeperQuorum ='ip:port[,ip:port]',
     tableName ='tableName',
-    rowKey ='colFamily:colName[,colFamily:colName]',
+    rowKey ='colName[,colName]',
     parallelism ='1',
     zookeeperParent ='/hbase'
  )
@@ -40,16 +40,16 @@ hbase2.0
 
 ## 5.样例：
 ```
-CREATE TABLE MyResult(
-    cf:channel varchar,
-    cf:pv BIGINT
+
+ CREATE TABLE MyResult(
+    cf:info VARCHAR,
+    cf:name VARCHAR,
+    cf:channel varchar
  )WITH(
     type ='hbase',
-    zookeeperQuorum ='rdos1:2181',
-    tableName ='workerinfo',
-    rowKey ='cf:channel',
-    parallelism ='1',
-    zookeeperParent ='/hbase'
- )
-
+    zookeeperQuorum ='xx:2181',
+    zookeeperParent ='/hbase',
+    tableName ='workerinfo01',
+    rowKey ='channel'
+ );
  ```

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java

@@ -38,6 +38,8 @@
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 /**
  * author: jingzhen@dtstack.com
@@ -53,6 +55,7 @@ public class HbaseOutputFormat extends MetricOutputFormat {
     private String tableName;
     private String[] columnNames;
     private String[] columnTypes;
+    private Map<String,String> columnNameFamily;
 
     private String[] families;
     private String[] qualifiers;
@@ -196,6 +199,11 @@ public HbaseOutputFormatBuilder setColumnTypes(String[] columnTypes) {
             return this;
         }
 
+        public HbaseOutputFormatBuilder setColumnNameFamily(Map<String, String> columnNameFamily) {
+            format.columnNameFamily = columnNameFamily;
+            return this;
+        }
+
         public HbaseOutputFormat finish() {
             Preconditions.checkNotNull(format.host, "zookeeperQuorum should be specified");
             Preconditions.checkNotNull(format.tableName, "tableName should be specified");
@@ -205,13 +213,16 @@ public HbaseOutputFormat finish() {
             String[] families = new String[format.columnNames.length];
             String[] qualifiers = new String[format.columnNames.length];
 
-            for(int i = 0; i < format.columnNames.length; ++i) {
-                String col = format.columnNames[i];
-                String[] part = col.split(":");
-                families[i] = part[0];
-                qualifiers[i] = part[1];
+            if (format.columnNameFamily != null) {
+                Set<String> keySet = format.columnNameFamily.keySet();
+                String[] columns = keySet.toArray(new String[keySet.size()]);
+                for (int i = 0; i < columns.length; ++i) {
+                    String col = columns[i];
+                    String[] part = col.split(":");
+                    families[i] = part[0];
+                    qualifiers[i] = part[1];
+                }
             }
-
             format.families = families;
             format.qualifiers = qualifiers;
 

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseSink.java

@@ -35,6 +35,8 @@
 import org.apache.flink.table.sinks.TableSink;
 import org.apache.flink.types.Row;
 
+import java.util.Map;
+
 /**
  * Date: 2018/09/14
  * Company: www.dtstack.com
@@ -43,6 +45,7 @@
 public class HbaseSink implements RetractStreamTableSink<Row>, IStreamSinkGener<HbaseSink> {
 
     protected String[] fieldNames;
+    protected Map<String, String> columnNameFamily;
     TypeInformation<?>[] fieldTypes;
     protected String zookeeperQuorum;
     protected String port;
@@ -62,6 +65,7 @@ public HbaseSink genStreamSink(TargetTableInfo targetTableInfo) {
         this.parent = hbaseTableInfo.getParent();
         this.tableName = hbaseTableInfo.getTableName();
         this.rowkey = hbaseTableInfo.getRowkey();
+        this.columnNameFamily = hbaseTableInfo.getColumnNameFamily();
         return this;
     }
 
@@ -72,6 +76,7 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
 
         builder.setRowkey(rowkey);
         builder.setColumnNames(fieldNames);
+        builder.setColumnNameFamily(columnNameFamily);
 
         HbaseOutputFormat outputFormat = builder.finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(outputFormat);

hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/table/HbaseSinkParser.java

@@ -23,8 +23,11 @@
 
 import com.dtstack.flink.sql.table.AbsTableParser;
 import com.dtstack.flink.sql.table.TableInfo;
+import com.dtstack.flink.sql.util.DtStringUtil;
 import com.dtstack.flink.sql.util.MathUtil;
 
+import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 
 import static com.dtstack.flink.sql.table.TableInfo.PARALLELISM_KEY;
@@ -65,4 +68,38 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         hbaseTableInfo.setRowkey(rk.split(","));
         return hbaseTableInfo;
     }
+
+    public void parseFieldsInfo(String fieldsInfo, HbaseTableInfo tableInfo){
+        List<String> fieldRows = DtStringUtil.splitIgnoreQuota(fieldsInfo, ',');
+        Map<String, String> columnFamilies = new LinkedHashMap<>();
+        for(String fieldRow : fieldRows){
+            fieldRow = fieldRow.trim();
+
+            String[] filedInfoArr = fieldRow.split("\\s+");
+            if(filedInfoArr.length < 2 ){
+                throw new RuntimeException(String.format("table [%s] field [%s] format error.", tableInfo.getName(), fieldRow));
+            }
+
+            boolean isMatcherKey = dealKeyPattern(fieldRow, tableInfo);
+            if(isMatcherKey){
+                continue;
+            }
+
+            //Compatible situation may arise in space in the fieldName
+            String[] filedNameArr = new String[filedInfoArr.length - 1];
+            System.arraycopy(filedInfoArr, 0, filedNameArr, 0, filedInfoArr.length - 1);
+            String fieldName = String.join(" ", filedNameArr);
+            String fieldType = filedInfoArr[filedInfoArr.length - 1 ].trim();
+            Class fieldClass = dbTypeConvertToJavaType(fieldType);
+            String[] columnFamily = fieldName.trim().split(":");
+            columnFamilies.put(fieldName.trim(),columnFamily[1]);
+            tableInfo.addPhysicalMappings(filedInfoArr[0],filedInfoArr[0]);
+            tableInfo.addField(columnFamily[1]);
+            tableInfo.addFieldClass(fieldClass);
+            tableInfo.addFieldType(fieldType);
+            tableInfo.addFieldExtraInfo(null);
+        }
+        tableInfo.setColumnNameFamily(columnFamilies);
+        tableInfo.finish();
+    }
 }