Merge branch 'revert-8fc47a79' into 'v1.8.0_dev'

Revert "Merge branch '1.8_dev_hbase_sink' into 'v1.8.0_dev'"

This reverts merge request !154

See merge request !156

Author: todd5167

docs/hbaseSink.md

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

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

@@ -38,8 +38,6 @@
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
-import java.util.Set;
 
 /**
  * author: jingzhen@dtstack.com
@@ -55,7 +53,6 @@ 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;
@@ -199,11 +196,6 @@ 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");
@@ -213,16 +205,13 @@ public HbaseOutputFormat finish() {
             String[] families = new String[format.columnNames.length];
             String[] qualifiers = new String[format.columnNames.length];
 
-            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];
-                }
+            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];
             }
+
             format.families = families;
             format.qualifiers = qualifiers;
 

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

@@ -35,8 +35,6 @@
 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
@@ -45,7 +43,6 @@
 public class HbaseSink implements RetractStreamTableSink<Row>, IStreamSinkGener<HbaseSink> {
 
     protected String[] fieldNames;
-    protected Map<String, String> columnNameFamily;
     TypeInformation<?>[] fieldTypes;
     protected String zookeeperQuorum;
     protected String port;
@@ -65,7 +62,6 @@ public HbaseSink genStreamSink(TargetTableInfo targetTableInfo) {
         this.parent = hbaseTableInfo.getParent();
         this.tableName = hbaseTableInfo.getTableName();
         this.rowkey = hbaseTableInfo.getRowkey();
-        this.columnNameFamily = hbaseTableInfo.getColumnNameFamily();
         return this;
     }
 
@@ -76,7 +72,6 @@ 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,11 +23,8 @@
 
 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;
@@ -68,38 +65,4 @@ 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();
-    }
 }