You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gobblin.apache.org by su...@apache.org on 2020/02/03 04:57:12 UTC
[incubator-gobblin] branch master updated:
[GOBBLIN-1035][Gobblin-1035] make hive dataset descriptor accepts regexed
db and tables
This is an automated email from the ASF dual-hosted git repository.
suvasude pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-gobblin.git
The following commit(s) were added to refs/heads/master by this push:
new ef3003c [GOBBLIN-1035][Gobblin-1035] make hive dataset descriptor accepts regexed db and tables
ef3003c is described below
commit ef3003c85531aaeacb760a4e30fca2d3069001aa
Author: Arjun <ab...@linkedin.com>
AuthorDate: Sun Feb 2 20:57:04 2020 -0800
[GOBBLIN-1035][Gobblin-1035] make hive dataset descriptor accepts regexed db and tables
Closes #2877 from
arjun4084346/regexedHiveWhitelist
---
.../gobblin/metrics/event/lineage/LineageInfo.java | 2 +-
.../modules/dataset/HiveDatasetDescriptor.java | 51 ++++++++++++----------
.../modules/dataset/HiveDatasetDescriptorTest.java | 35 +++++++++++++--
3 files changed, 59 insertions(+), 29 deletions(-)
diff --git a/gobblin-metrics-libs/gobblin-metrics-base/src/main/java/org/apache/gobblin/metrics/event/lineage/LineageInfo.java b/gobblin-metrics-libs/gobblin-metrics-base/src/main/java/org/apache/gobblin/metrics/event/lineage/LineageInfo.java
index ed8903d..cd1f373 100644
--- a/gobblin-metrics-libs/gobblin-metrics-base/src/main/java/org/apache/gobblin/metrics/event/lineage/LineageInfo.java
+++ b/gobblin-metrics-libs/gobblin-metrics-base/src/main/java/org/apache/gobblin/metrics/event/lineage/LineageInfo.java
@@ -148,7 +148,7 @@ public final class LineageInfo {
return;
}
- log.info(String.format("Put destination %s for branch %d", Descriptor.toJson(descriptors), branchId));
+ log.debug(String.format("Put destination %s for branch %d", Descriptor.toJson(descriptors), branchId));
synchronized (state.getProp(getKey(NAME_KEY))) {
List<Descriptor> resolvedDescriptors = new ArrayList<>();
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/dataset/HiveDatasetDescriptor.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/dataset/HiveDatasetDescriptor.java
index ce28efb..fb1bcd4 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/dataset/HiveDatasetDescriptor.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/dataset/HiveDatasetDescriptor.java
@@ -19,7 +19,6 @@ package org.apache.gobblin.service.modules.dataset;
import java.io.IOException;
import java.util.List;
-import java.util.regex.Pattern;
import org.apache.hadoop.fs.GlobPattern;
@@ -28,29 +27,32 @@ import com.typesafe.config.Config;
import com.typesafe.config.ConfigValueFactory;
import lombok.EqualsAndHashCode;
+import lombok.extern.slf4j.Slf4j;
import org.apache.gobblin.data.management.copy.hive.HiveCopyEntityHelper;
+import org.apache.gobblin.data.management.copy.hive.HiveDatasetFinder;
+import org.apache.gobblin.data.management.copy.hive.WhitelistBlacklist;
import org.apache.gobblin.data.management.version.finder.DatePartitionHiveVersionFinder;
import org.apache.gobblin.util.ConfigUtils;
-import org.apache.gobblin.util.PathUtils;
+@Slf4j
/**
* As of now, {@link HiveDatasetDescriptor} has same implementation as that of {@link SqlDatasetDescriptor}.
* Fields {@link HiveDatasetDescriptor#isPartitioned}, {@link HiveDatasetDescriptor#partitionColumn} and
* {@link HiveDatasetDescriptor#partitionFormat} are used for methods 'equals' and 'hashCode'.
*/
-@EqualsAndHashCode (callSuper = true)
+@EqualsAndHashCode (exclude = {"whitelistBlacklist"}, callSuper = true)
public class HiveDatasetDescriptor extends SqlDatasetDescriptor {
static final String IS_PARTITIONED_KEY = "isPartitioned";
static final String PARTITION_COLUMN = "partition.column";
static final String PARTITION_FORMAT = "partition.format";
static final String CONFLICT_POLICY = "conflict.policy";
- static final String WHITELIST_TABLES = "whitelist.tables";
private final boolean isPartitioned;
private final String partitionColumn;
private final String partitionFormat;
private final String conflictPolicy;
+ WhitelistBlacklist whitelistBlacklist;
public HiveDatasetDescriptor(Config config) throws IOException {
super(config);
@@ -65,16 +67,30 @@ public class HiveDatasetDescriptor extends SqlDatasetDescriptor {
partitionFormat = "";
conflictPolicy = HiveCopyEntityHelper.ExistingEntityPolicy.REPLACE_TABLE.name();
}
+
+ whitelistBlacklist = new WhitelistBlacklist(config.withValue(WhitelistBlacklist.WHITELIST,
+ ConfigValueFactory.fromAnyRef(createHiveDatasetWhitelist())));
this.setRawConfig(this.getRawConfig()
.withValue(CONFLICT_POLICY, ConfigValueFactory.fromAnyRef(conflictPolicy))
.withValue(PARTITION_COLUMN, ConfigValueFactory.fromAnyRef(partitionColumn))
.withValue(PARTITION_FORMAT, ConfigValueFactory.fromAnyRef(partitionFormat))
- .withValue(WHITELIST_TABLES, ConfigValueFactory.fromAnyRef(createWhitelistedTables())
+ .withValue(HiveDatasetFinder.HIVE_DATASET_PREFIX + "." + WhitelistBlacklist.WHITELIST,
+ ConfigValueFactory.fromAnyRef(createHiveDatasetWhitelist())
));
}
- private String createWhitelistedTables() {
- return this.tableName.replace(',', '|');
+ // Using Hadoop's GlobPattern instead of java.util.regex, because could not find any API in java.util.regex
+ // which tells if the string is a plain string or contains special characters.
+ // If the db name contains wildcards, whitelist is created as <regex_db>.*
+ // Otherwise, whitelist is created as <db>.tables.
+ // This is the format which HiveDatasetFinder understands.
+ // e.g. db=testDb, table=foo*,bar*, whitelist will be testDb.foo*|bar*
+ String createHiveDatasetWhitelist() {
+ if (new GlobPattern(this.databaseName).hasWildcard()) {
+ return this.databaseName + ".*";
+ } else {
+ return this.databaseName + "." + this.tableName.replace(',', '|');
+ }
}
@Override
@@ -102,29 +118,16 @@ public class HiveDatasetDescriptor extends SqlDatasetDescriptor {
String otherDbName = parts.get(0);
String otherTableNames = parts.get(1);
- if(!Pattern.compile(this.databaseName).matcher(otherDbName).matches()) {
+ if (!this.whitelistBlacklist.acceptDb(otherDbName)) {
return false;
}
- List<String> tables = Splitter.on(",").splitToList(otherTableNames);
- for (String table : tables) {
- if (!isPathContaining(table)) {
+ List<String> otherTables = Splitter.on(",").splitToList(otherTableNames);
+ for (String otherTable : otherTables) {
+ if (!this.whitelistBlacklist.acceptTable(otherDbName, otherTable)) {
return false;
}
}
return true;
}
-
- private boolean isPathContaining(String tableName) {
- if (tableName == null) {
- return false;
- }
-
- if (PathUtils.GLOB_TOKENS.matcher(tableName).find()) {
- return false;
- }
-
- GlobPattern globPattern = new GlobPattern(this.tableName);
- return globPattern.matches(tableName);
- }
}
diff --git a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/dataset/HiveDatasetDescriptorTest.java b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/dataset/HiveDatasetDescriptorTest.java
index 5c953b6..21e4410 100644
--- a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/dataset/HiveDatasetDescriptorTest.java
+++ b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/dataset/HiveDatasetDescriptorTest.java
@@ -30,16 +30,43 @@ import org.apache.gobblin.service.modules.flowgraph.DatasetDescriptorConfigKeys;
public class HiveDatasetDescriptorTest {
+ Config baseConfig = ConfigFactory.empty().withValue(DatasetDescriptorConfigKeys.PLATFORM_KEY, ConfigValueFactory.fromAnyRef("hive"))
+ .withValue(DatasetDescriptorConfigKeys.DATABASE_KEY, ConfigValueFactory.fromAnyRef("testDb_Db1"))
+ .withValue(DatasetDescriptorConfigKeys.TABLE_KEY, ConfigValueFactory.fromAnyRef("testTable_Table1"));;
@Test
public void objectCreation() throws IOException {
- Config baseConfig = ConfigFactory.empty().withValue(DatasetDescriptorConfigKeys.PLATFORM_KEY, ConfigValueFactory.fromAnyRef("hive"))
- .withValue(DatasetDescriptorConfigKeys.DATABASE_KEY, ConfigValueFactory.fromAnyRef("testDb_Db1"))
- .withValue(DatasetDescriptorConfigKeys.TABLE_KEY, ConfigValueFactory.fromAnyRef("testTable_Table1"));;
-
SqlDatasetDescriptor descriptor1 = new HiveDatasetDescriptor(baseConfig.withValue(HiveDatasetDescriptor.IS_PARTITIONED_KEY, ConfigValueFactory.fromAnyRef(true)));
SqlDatasetDescriptor descriptor2 = new HiveDatasetDescriptor(baseConfig.withValue(HiveDatasetDescriptor.IS_PARTITIONED_KEY, ConfigValueFactory.fromAnyRef(false)));
Assert.assertNotEquals(descriptor1, descriptor2);
}
+
+ @Test
+ public void testWhitelist() throws IOException {
+ HiveDatasetDescriptor descriptor = new HiveDatasetDescriptor(baseConfig
+ .withValue(DatasetDescriptorConfigKeys.DATABASE_KEY, ConfigValueFactory.fromAnyRef("test*"))
+ .withValue(DatasetDescriptorConfigKeys.TABLE_KEY, ConfigValueFactory.fromAnyRef("abc,def,fgh"))
+ );
+ Assert.assertTrue(descriptor.whitelistBlacklist.acceptTable("testDb", "abc"));
+ Assert.assertTrue(descriptor.whitelistBlacklist.acceptTable("testDb", "abca"));
+ Assert.assertFalse(descriptor.whitelistBlacklist.acceptTable("otherTestDb", "abc"));
+
+ descriptor = new HiveDatasetDescriptor(baseConfig
+ .withValue(DatasetDescriptorConfigKeys.DATABASE_KEY, ConfigValueFactory.fromAnyRef("testDb"))
+ .withValue(DatasetDescriptorConfigKeys.TABLE_KEY, ConfigValueFactory.fromAnyRef("abc,def,ghi"))
+ );
+ Assert.assertTrue(descriptor.whitelistBlacklist.acceptTable("testDb", "abc"));
+ Assert.assertFalse(descriptor.whitelistBlacklist.acceptTable("testDb", "abca"));
+ Assert.assertFalse(descriptor.whitelistBlacklist.acceptTable("otherTestDb", "abc"));
+
+ descriptor = new HiveDatasetDescriptor(baseConfig
+ .withValue(DatasetDescriptorConfigKeys.DATABASE_KEY, ConfigValueFactory.fromAnyRef("testDb"))
+ .withValue(DatasetDescriptorConfigKeys.TABLE_KEY, ConfigValueFactory.fromAnyRef("abc*,ghi"))
+ );
+ Assert.assertTrue(descriptor.whitelistBlacklist.acceptTable("testDb", "abc"));
+ Assert.assertTrue(descriptor.whitelistBlacklist.acceptTable("testDb", "ghi"));
+ Assert.assertTrue(descriptor.whitelistBlacklist.acceptTable("testDb", "abcabc"));
+ Assert.assertFalse(descriptor.whitelistBlacklist.acceptTable("otherTestDb", "abc"));
+ }
}
\ No newline at end of file