Merge branch 'feat_1.8_rowkey' into 'v1.8.0_dev'

hbase 结果表支持 rowKey

See merge request dt-insight-engine/flinkStreamSQL!15

Author: zoudaokoulife

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

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+ 
+
+package com.dtstack.flink.sql.sink.hbase;
+
+
+import com.dtstack.flink.sql.sink.hbase.enums.EReplaceOpType;
+
+import java.util.Map;
+
+/**
+ * Reason:
+ * Date: 2018/8/23
+ * Company: www.dtstack.com
+ * @author xuchao
+ */
+
+public abstract class AbstractReplaceOperator {
+
+    private EReplaceOpType opType;
+
+    public AbstractReplaceOperator(EReplaceOpType opType){
+        this.opType = opType;
+    }
+
+    public String doOperator(Map<String, Object> refData){
+        String replaceStr = replaceStr(refData);
+        return doFunc(replaceStr);
+    }
+
+    public String replaceStr(Map<String, Object> refData){
+        return "";
+    }
+
+    /**
+     * The processing function to provide custom
+     * @param replaceStr
+     * @return
+     */
+    abstract String doFunc(String replaceStr);
+}

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

@@ -23,6 +23,7 @@
 import com.dtstack.flink.sql.enums.EUpdateMode;
 import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
@@ -40,7 +41,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.text.SimpleDateFormat;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -55,7 +55,7 @@ public class HbaseOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
 
     private String host;
     private String zkParent;
-    private String[] rowkey;
+    private String rowkey;
     private String tableName;
     private String[] columnNames;
     private String updateMode;
@@ -69,8 +69,6 @@ public class HbaseOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
     private transient Connection conn;
     private transient Table table;
 
-    public final SimpleDateFormat ROWKEY_DATE_FORMAT = new SimpleDateFormat("yyyyMMddHHmmss");
-
     @Override
     public void configure(Configuration parameters) {
         LOG.warn("---configure---");
@@ -166,40 +164,29 @@ private Put getPutByRow(Row record) {
     }
 
     private String buildRowKey(Row record) {
-        List<String> rowKeyValues = getRowKeyValues(record);
+        String rowKeyValues = getRowKeyValues(record);
         // all rowkey not null
-        if (rowKeyValues.size() != rowkey.length) {
+        if (StringUtils.isBlank(rowKeyValues)) {
             LOG.error("row key value must not null,record is ..{}", record);
             outDirtyRecords.inc();
             return "";
         }
-        return StringUtils.join(rowKeyValues, "-");
+        return rowKeyValues;
     }
 
-    private List<String> getRowKeyValues(Row record) {
-        List<String> rowKeyValues = Lists.newArrayList();
-        for (int i = 0; i < rowkey.length; ++i) {
-            String colName = rowkey[i];
-            int rowKeyIndex = 0;
-            for (; rowKeyIndex < columnNames.length; ++rowKeyIndex) {
-                if (columnNames[rowKeyIndex].equals(colName)) {
-                    break;
-                }
-            }
+    private String getRowKeyValues(Row record) {
+        Map<String, Object> row = rowConvertMap(record);
+        RowKeyBuilder rowKeyBuilder = new RowKeyBuilder();
+        rowKeyBuilder.init(rowkey);
+        return rowKeyBuilder.getRowKey(row);
+    }
 
-            if (rowKeyIndex != columnNames.length && record.getField(rowKeyIndex) != null) {
-                Object field = record.getField(rowKeyIndex);
-                if (field == null) {
-                    continue;
-                } else if (field instanceof java.util.Date) {
-                    java.util.Date d = (java.util.Date) field;
-                    rowKeyValues.add(ROWKEY_DATE_FORMAT.format(d));
-                } else {
-                    rowKeyValues.add(field.toString());
-                }
-            }
+    private Map<String, Object> rowConvertMap(Row record){
+        Map<String, Object> rowValue = Maps.newHashMap();
+        for(int i = 0; i < columnNames.length; i++){
+            rowValue.put(columnNames[i], record.getField(i));
         }
-        return rowKeyValues;
+        return rowValue;
     }
 
     @Override
@@ -241,7 +228,7 @@ public HbaseOutputFormatBuilder setTable(String tableName) {
             return this;
         }
 
-        public HbaseOutputFormatBuilder setRowkey(String[] rowkey) {
+        public HbaseOutputFormatBuilder setRowkey(String rowkey) {
             format.rowkey = rowkey;
             return this;
         }

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

@@ -51,7 +51,7 @@ public class HbaseSink implements RetractStreamTableSink<Row>, IStreamSinkGener<
     protected String parent;
     protected String tableName;
     protected String updateMode;
-    protected String[] rowkey;
+    protected String rowkey;
 
     public HbaseSink() {
         // TO DO NOTHING

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

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+ 
+
+package com.dtstack.flink.sql.sink.hbase;
+
+import com.dtstack.flink.sql.sink.hbase.enums.EReplaceOpType;
+import com.dtstack.flink.sql.util.MD5Utils;
+
+/**
+ * Reason:
+ * Date: 2018/8/23
+ * Company: www.dtstack.com
+ * @author xuchao
+ */
+
+public class Md5ReplaceOperator extends AbstractReplaceOperator {
+
+    public Md5ReplaceOperator(EReplaceOpType opType) {
+        super(opType);
+    }
+
+    @Override
+    String doFunc(String replaceStr) {
+        return MD5Utils.getMD5String(replaceStr);
+    }
+}

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

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+ 
+
+package com.dtstack.flink.sql.sink.hbase;
+
+
+import com.dtstack.flink.sql.sink.hbase.enums.EReplaceType;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * Reason:
+ * Date: 2018/8/23
+ * Company: www.dtstack.com
+ *
+ * @author xuchao
+ */
+
+public class ReplaceInfo implements Serializable {
+
+    private static final long serialVersionUID = 2058635242957737717L;
+
+    private EReplaceType type;
+
+    private String param;
+    
+    private List<ReplaceInfo> subReplaceInfos;
+
+    public ReplaceInfo(EReplaceType type){
+        this.type = type;
+    }
+
+    public EReplaceType getType() {
+        return type;
+    }
+
+    public void setType(EReplaceType type) {
+        this.type = type;
+    }
+
+    public String getParam() {
+        return param;
+    }
+
+    public void setParam(String param) {
+        this.param = param;
+    }
+    
+    public List<ReplaceInfo> getSubReplaceInfos() {
+        return subReplaceInfos;
+    }
+
+    public void setSubReplaceInfos(List<ReplaceInfo> subReplaceInfos) {
+        this.subReplaceInfos = subReplaceInfos;
+    }
+}