Add pattern option for custom transformations from kafka topic names …

…to hive tables names #155
confluentinc · Dec 29, 2018 · 10b5085 · 10b5085
1 parent 40aaa1f
commit 10b5085
Show file tree

Hide file tree

Showing 2 changed files with 39 additions and 1 deletion.
diff --git a/hive/src/main/java/io/confluent/connect/storage/hive/HiveConfig.java b/hive/src/main/java/io/confluent/connect/storage/hive/HiveConfig.java
@@ -60,6 +60,13 @@ public class HiveConfig extends AbstractConfig implements ComposableConfig {
   public static final String HIVE_DATABASE_DEFAULT = "default";
   public static final String HIVE_DATABASE_DISPLAY = "Hive database";
 
+  public static final String HIVE_TABLE_PATTERN_CONFIG = "hive.table.pattern";
+  public static final String HIVE_TABLE_PATTERN_DOC =
+      "Regular expression for transformation from Kafka topic name to Hive table name. "
+      + "If regular expression doesn't match, Kafka topic name will be used as Hive table name.";
+  public static final String HIVE_TABLE_PATTERN_DEFAULT = "";
+  public static final String HIVE_TABLE_PATTERN_DISPLAY = "";
+
   // Schema group
   public static final String SCHEMA_COMPATIBILITY_CONFIG = "schema.compatibility";
   public static final String SCHEMA_COMPATIBILITY_DOC =
@@ -153,6 +160,18 @@ public class HiveConfig extends AbstractConfig implements ComposableConfig {
           HIVE_DATABASE_DISPLAY,
           hiveIntegrationDependentsRecommender
       );
+
+      CONFIG_DEF.define(
+              HIVE_TABLE_PATTERN_CONFIG,
+          Type.STRING,
+              HIVE_TABLE_PATTERN_DEFAULT,
+          Importance.LOW,
+              HIVE_TABLE_PATTERN_DOC,
+          group,
+          ++orderInGroup,
+          Width.SHORT,
+              HIVE_TABLE_PATTERN_DISPLAY
+      );
     }
 
     {

diff --git a/hive/src/main/java/io/confluent/connect/storage/hive/HiveMetaStore.java b/hive/src/main/java/io/confluent/connect/storage/hive/HiveMetaStore.java
@@ -37,13 +37,16 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import io.confluent.connect.storage.errors.HiveMetaStoreException;
 
 public class HiveMetaStore {
 
   private static final Logger log = LoggerFactory.getLogger(HiveMetaStore.class);
   protected final IMetaStoreClient client;
+  private Pattern hiveTableMatcher;
 
   public HiveMetaStore(AbstractConfig connectorConfig) {
     this(new Configuration(), connectorConfig);
@@ -78,6 +81,10 @@ public HiveMetaStore(Configuration conf, AbstractConfig connectorConfig)
     } catch (IOException | MetaException e) {
       throw new HiveMetaStoreException(e);
     }
+
+    String hiveTablePattern = connectorConfig
+        .getString(HiveConfig.HIVE_TABLE_PATTERN_CONFIG);
+    hiveTableMatcher = Pattern.compile(hiveTablePattern);
   }
 
   private interface ClientAction<R> {
@@ -364,6 +371,18 @@ public List<String> call() throws TException {
   }
 
   public String tableNameConverter(String table) {
-    return table == null ? table : table.replaceAll("[.-]", "_");
+    if (table == null) {
+      return null;
+    }
+
+    String converterTable;
+    Matcher tableMatcher = hiveTableMatcher.matcher(table);
+    if (tableMatcher.find() && tableMatcher.groupCount() == 1) {
+      converterTable = tableMatcher.group(1);
+    } else {
+      converterTable = table;
+    }
+
+    return converterTable.replaceAll("[.-]", "_");
   }
 }