hbase 结果表支持 rowKey

Author: dapeng

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;
@@ -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---");
@@ -178,30 +176,23 @@ private String buildRowKey(Row record) {
 
     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;
-                }
-            }
-
-            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());
-                }
-            }
+        Map<String, Object> row = rowConvertMap(record);
+        for (String key : rowkey) {
+            RowKeyBuilder rowKeyBuilder = new RowKeyBuilder();
+            rowKeyBuilder.init(key);
+           rowKeyValues.add(rowKeyBuilder.getRowKey(row));
         }
         return rowKeyValues;
     }
 
+    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 rowValue;
+    }
+
     @Override
     public void close() throws IOException {
         if (conn != null) {

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;
+    }
+}

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

@@ -0,0 +1,148 @@
+/*
+ * 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 com.dtstack.flink.sql.util.MD5Utils;
+import com.google.common.collect.Lists;
+import org.apache.commons.collections.CollectionUtils;
+
+import javax.xml.crypto.Data;
+import java.io.Serializable;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * rowkey rule
+ * Date: 2018/8/23
+ * Company: www.dtstack.com
+ * @author xuchao
+ */
+
+public class RowKeyBuilder implements Serializable{
+
+    private static final long serialVersionUID = 2058635242857937717L;
+
+    private static Pattern Md5Operator = Pattern.compile("(?i)^md5\\(\\s*(.*)\\s*\\)$");
+
+    private List<ReplaceInfo> operatorChain = Lists.newArrayList();
+
+    public void init(String rowKeyTempl){
+    	operatorChain.addAll(makeFormula(rowKeyTempl));
+    }
+
+    /**
+     *
+     * @param refData
+     * @return
+     */
+    public String getRowKey(Map<String, Object> refData){
+    	return buildStr(operatorChain, refData);
+    }
+
+
+
+    private String buildStr(List<ReplaceInfo> fieldList, Map<String, Object> refData){
+        if(CollectionUtils.isEmpty(fieldList)){
+            return "";
+        }
+        StringBuffer sb = new StringBuffer("");
+        for(ReplaceInfo replaceInfo : fieldList){
+
+            if(replaceInfo.getType() == EReplaceType.CONSTANT){
+                sb.append(replaceInfo.getParam());
+                continue;
+            }
+
+            if(replaceInfo.getType() == EReplaceType.FUNC){
+                sb.append(MD5Utils.getMD5String(buildStr(replaceInfo.getSubReplaceInfos(), refData)));
+                continue;
+            }
+            String replaceName = replaceInfo.getParam();
+            if(!refData.containsKey(replaceName)){
+                throw new RuntimeException(String.format("build rowKey with field %s which value not found.", replaceName));
+            }
+            Object value = refData.get(replaceName);
+            if(value instanceof Date){
+                SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
+                java.util.Date d = (java.util.Date) value;
+                sb.append(sdf.format(d));
+            } else {
+                sb.append(value);
+            }
+        }
+
+        return sb.toString();
+    }
+
+    public static String[] splitIgnoreQuotaBrackets(String str, String delimiter){
+        String splitPatternStr = delimiter + "(?![^()]*+\\))(?![^{}]*+})(?![^\\[\\]]*+\\])";
+        return str.split(splitPatternStr);
+    }
+
+    /**
+     *
+     * @param field
+     * @return
+     */
+    public ReplaceInfo getReplaceInfo(String field){
+
+        field = field.trim();
+        if(field.length() <= 2){
+            throw new RuntimeException(field + " \n" +
+                    "Format defined exceptions");
+        }
+
+        //判断是不是常量==>''包裹的标识
+        if(field.startsWith("'") && field.endsWith("'")){
+            ReplaceInfo replaceInfo = new ReplaceInfo(EReplaceType.CONSTANT);
+            field = field.substring(1, field.length() - 1);
+            replaceInfo.setParam(field);
+            return replaceInfo;
+        }
+
+        ReplaceInfo replaceInfo = new ReplaceInfo(EReplaceType.PARAM);
+        replaceInfo.setParam(field);
+        return replaceInfo;
+    }
+
+    private List<ReplaceInfo> makeFormula(String formula){
+        if(formula == null || formula.length() <= 0){
+            Lists.newArrayList();
+        }
+        List<ReplaceInfo> result = Lists.newArrayList();
+        for(String meta: splitIgnoreQuotaBrackets(formula, "\\+")){
+            Matcher matcher = Md5Operator.matcher(meta.trim());
+            if(matcher.find()){
+                ReplaceInfo replaceInfo = new ReplaceInfo(EReplaceType.FUNC);
+                replaceInfo.setSubReplaceInfos(makeFormula(matcher.group(1)));
+                result.add(replaceInfo);
+            } else {
+                result.add(getReplaceInfo(meta));
+            }
+        }
+        return result;
+    }
+}