rowkey的字段拼接形式修改

Author: dapeng

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

@@ -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;
@@ -164,25 +164,21 @@ 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();
+    private String getRowKeyValues(Row record) {
         Map<String, Object> row = rowConvertMap(record);
-        for (String key : rowkey) {
-            RowKeyBuilder rowKeyBuilder = new RowKeyBuilder();
-            rowKeyBuilder.init(key);
-           rowKeyValues.add(rowKeyBuilder.getRowKey(row));
-        }
-        return rowKeyValues;
+        RowKeyBuilder rowKeyBuilder = new RowKeyBuilder();
+        rowKeyBuilder.init(rowkey);
+        return rowKeyBuilder.getRowKey(row);
     }
 
     private Map<String, Object> rowConvertMap(Row record){
@@ -232,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/table/HbaseSinkParser.java

@@ -69,7 +69,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         hbaseTableInfo.setHost((String) props.get(HBASE_ZOOKEEPER_QUORUM.toLowerCase()));
         hbaseTableInfo.setParent((String)props.get(ZOOKEEPER_PARENT.toLowerCase()));
         String rk = (String) props.get(HBASE_ROWKEY.toLowerCase());
-        hbaseTableInfo.setRowkey(StringUtils.split(rk, ","));
+        hbaseTableInfo.setRowkey(rk);
         String updateMode = (String) props.getOrDefault(UPDATE_KEY, EUpdateMode.APPEND.name());
         hbaseTableInfo.setUpdateMode(updateMode);
         return hbaseTableInfo;

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

@@ -40,7 +40,7 @@ public class HbaseTableInfo extends AbstractTargetTableInfo {
 
     private String parent;
 
-    private String[] rowkey;
+    private String rowkey;
 
     private Map<String, String> columnNameFamily;
 
@@ -90,11 +90,11 @@ public void setParent(String parent) {
         this.parent = parent;
     }
 
-    public String[] getRowkey() {
+    public String getRowkey() {
         return rowkey;
     }
 
-    public void setRowkey(String[] rowkey) {
+    public void setRowkey(String rowkey) {
         this.rowkey = rowkey;
     }