You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by go...@apache.org on 2020/09/30 02:32:08 UTC
[flink] branch master updated: [FLINK-19292][hive] HiveCatalog
should support specifying Hadoop conf dir with configuration (#13434)
This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new b4ae117 [FLINK-19292][hive] HiveCatalog should support specifying Hadoop conf dir with configuration (#13434)
b4ae117 is described below
commit b4ae1170b11cabcb6737c390508e8b8cdfc7c6bd
Author: SteNicholas <pr...@163.com>
AuthorDate: Wed Sep 30 10:30:46 2020 +0800
[FLINK-19292][hive] HiveCatalog should support specifying Hadoop conf dir with configuration (#13434)
---
.../flink/table/catalog/hive/HiveCatalog.java | 62 ++++++++++++----------
.../hive/descriptors/HiveCatalogValidator.java | 2 +
.../catalog/hive/factories/HiveCatalogFactory.java | 7 ++-
.../table/catalog/hive/util/HiveTableUtil.java | 34 ++++++++++++
.../flink/table/catalog/hive/HiveTestUtils.java | 5 ++
.../hive/factories/HiveCatalogFactoryTest.java | 28 +++++++++-
6 files changed, 109 insertions(+), 29 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
index 858feb9..768b28b 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
@@ -74,10 +74,8 @@ import org.apache.flink.table.factories.Factory;
import org.apache.flink.table.factories.FunctionDefinitionFactory;
import org.apache.flink.table.factories.TableFactory;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.StringUtils;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.FileUtils;
import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.conf.HiveConf;
@@ -103,7 +101,6 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
-import java.io.File;
import java.net.MalformedURLException;
import java.nio.file.Paths;
import java.util.ArrayList;
@@ -125,9 +122,11 @@ import static org.apache.flink.sql.parser.hive.ddl.SqlCreateHiveTable.PK_CONSTRA
import static org.apache.flink.table.catalog.config.CatalogConfig.FLINK_PROPERTY_PREFIX;
import static org.apache.flink.table.catalog.hive.util.HiveStatsUtil.parsePositiveIntStat;
import static org.apache.flink.table.catalog.hive.util.HiveStatsUtil.parsePositiveLongStat;
+import static org.apache.flink.table.catalog.hive.util.HiveTableUtil.getHadoopConfiguration;
import static org.apache.flink.table.utils.PartitionPathUtils.unescapePathName;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.StringUtils.isNullOrWhitespaceOnly;
/**
* A catalog implementation for Hive.
@@ -152,14 +151,22 @@ public class HiveCatalog extends AbstractCatalog {
HiveMetastoreClientWrapper client;
public HiveCatalog(String catalogName, @Nullable String defaultDatabase, @Nullable String hiveConfDir) {
- this(catalogName, defaultDatabase, hiveConfDir, HiveShimLoader.getHiveVersion());
+ this(catalogName, defaultDatabase, hiveConfDir, null);
}
public HiveCatalog(String catalogName, @Nullable String defaultDatabase, @Nullable String hiveConfDir, String hiveVersion) {
+ this(catalogName, defaultDatabase, hiveConfDir, null, hiveVersion);
+ }
+
+ public HiveCatalog(String catalogName, @Nullable String defaultDatabase, @Nullable String hiveConfDir, @Nullable String hadoopConfDir, @Nullable String hiveVersion) {
+ this(catalogName, defaultDatabase, createHiveConf(hiveConfDir, hadoopConfDir), hiveVersion);
+ }
+
+ public HiveCatalog(String catalogName, @Nullable String defaultDatabase, @Nullable HiveConf hiveConf, @Nullable String hiveVersion) {
this(catalogName,
defaultDatabase == null ? DEFAULT_DB : defaultDatabase,
- createHiveConf(hiveConfDir),
- hiveVersion,
+ hiveConf,
+ isNullOrWhitespaceOnly(hiveVersion) ? HiveShimLoader.getHiveVersion() : hiveVersion,
false);
}
@@ -168,13 +175,13 @@ public class HiveCatalog extends AbstractCatalog {
boolean allowEmbedded) {
super(catalogName, defaultDatabase == null ? DEFAULT_DB : defaultDatabase);
- this.hiveConf = hiveConf == null ? createHiveConf(null) : hiveConf;
+ this.hiveConf = hiveConf == null ? createHiveConf(null, null) : hiveConf;
if (!allowEmbedded) {
checkArgument(!isEmbeddedMetastore(this.hiveConf),
"Embedded metastore is not allowed. Make sure you have set a valid value for " +
HiveConf.ConfVars.METASTOREURIS.toString());
}
- checkArgument(!StringUtils.isNullOrWhitespaceOnly(hiveVersion), "hiveVersion cannot be null or empty");
+ checkArgument(!isNullOrWhitespaceOnly(hiveVersion), "hiveVersion cannot be null or empty");
this.hiveVersion = hiveVersion;
hiveShim = HiveShimLoader.loadHiveShim(hiveVersion);
// add this to hiveConf to make sure table factory and source/sink see the same Hive version as HiveCatalog
@@ -183,7 +190,7 @@ public class HiveCatalog extends AbstractCatalog {
LOG.info("Created HiveCatalog '{}'", catalogName);
}
- private static HiveConf createHiveConf(@Nullable String hiveConfDir) {
+ private static HiveConf createHiveConf(@Nullable String hiveConfDir, @Nullable String hadoopConfDir) {
LOG.info("Setting hive conf dir as {}", hiveConfDir);
try {
@@ -195,18 +202,19 @@ public class HiveCatalog extends AbstractCatalog {
String.format("Failed to get hive-site.xml from %s", hiveConfDir), e);
}
- // create HiveConf from hadoop configuration
- Configuration hadoopConf = HadoopUtils.getHadoopConfiguration(new org.apache.flink.configuration.Configuration());
-
- // Add mapred-site.xml. We need to read configurations like compression codec.
- for (String possibleHadoopConfPath : HadoopUtils.possibleHadoopConfPaths(new org.apache.flink.configuration.Configuration())) {
- File mapredSite = new File(new File(possibleHadoopConfPath), "mapred-site.xml");
- if (mapredSite.exists()) {
- hadoopConf.addResource(new Path(mapredSite.getAbsolutePath()));
- break;
+ // create HiveConf from hadoop configuration with hadoop conf directory configured.
+ Configuration hadoopConf = null;
+ if (isNullOrWhitespaceOnly(hadoopConfDir)) {
+ for (String possibleHadoopConfPath : HadoopUtils.possibleHadoopConfPaths(new org.apache.flink.configuration.Configuration())) {
+ hadoopConf = getHadoopConfiguration(possibleHadoopConfPath);
+ if (hadoopConf != null) {
+ break;
+ }
}
+ } else {
+ hadoopConf = getHadoopConfiguration(hadoopConfDir);
}
- return new HiveConf(hadoopConf, HiveConf.class);
+ return new HiveConf(hadoopConf == null ? new Configuration() : hadoopConf, HiveConf.class);
}
@VisibleForTesting
@@ -275,7 +283,7 @@ public class HiveCatalog extends AbstractCatalog {
@Override
public void createDatabase(String databaseName, CatalogDatabase database, boolean ignoreIfExists)
throws DatabaseAlreadyExistException, CatalogException {
- checkArgument(!StringUtils.isNullOrWhitespaceOnly(databaseName), "databaseName cannot be null or empty");
+ checkArgument(!isNullOrWhitespaceOnly(databaseName), "databaseName cannot be null or empty");
checkNotNull(database, "database cannot be null");
Database hiveDatabase = HiveDatabaseUtil.instantiateHiveDatabase(databaseName, database);
@@ -294,7 +302,7 @@ public class HiveCatalog extends AbstractCatalog {
@Override
public void alterDatabase(String databaseName, CatalogDatabase newDatabase, boolean ignoreIfNotExists)
throws DatabaseNotExistException, CatalogException {
- checkArgument(!StringUtils.isNullOrWhitespaceOnly(databaseName), "databaseName cannot be null or empty");
+ checkArgument(!isNullOrWhitespaceOnly(databaseName), "databaseName cannot be null or empty");
checkNotNull(newDatabase, "newDatabase cannot be null");
// client.alterDatabase doesn't throw any exception if there is no existing database
@@ -447,7 +455,7 @@ public class HiveCatalog extends AbstractCatalog {
public void renameTable(ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists)
throws TableNotExistException, TableAlreadyExistException, CatalogException {
checkNotNull(tablePath, "tablePath cannot be null");
- checkArgument(!StringUtils.isNullOrWhitespaceOnly(newTableName), "newTableName cannot be null or empty");
+ checkArgument(!isNullOrWhitespaceOnly(newTableName), "newTableName cannot be null or empty");
try {
// alter_table() doesn't throw a clear exception when target table doesn't exist.
@@ -547,7 +555,7 @@ public class HiveCatalog extends AbstractCatalog {
@Override
public List<String> listTables(String databaseName) throws DatabaseNotExistException, CatalogException {
- checkArgument(!StringUtils.isNullOrWhitespaceOnly(databaseName), "databaseName cannot be null or empty");
+ checkArgument(!isNullOrWhitespaceOnly(databaseName), "databaseName cannot be null or empty");
try {
return client.getAllTables(databaseName);
@@ -561,7 +569,7 @@ public class HiveCatalog extends AbstractCatalog {
@Override
public List<String> listViews(String databaseName) throws DatabaseNotExistException, CatalogException {
- checkArgument(!StringUtils.isNullOrWhitespaceOnly(databaseName), "databaseName cannot be null or empty");
+ checkArgument(!isNullOrWhitespaceOnly(databaseName), "databaseName cannot be null or empty");
try {
return client.getViews(databaseName);
@@ -915,7 +923,7 @@ public class HiveCatalog extends AbstractCatalog {
partitionSpec, partCols, new ObjectPath(hiveTable.getDbName(), hiveTable.getTableName()));
// validate partition values
for (int i = 0; i < partCols.size(); i++) {
- if (StringUtils.isNullOrWhitespaceOnly(partValues.get(i))) {
+ if (isNullOrWhitespaceOnly(partValues.get(i))) {
throw new PartitionSpecInvalidException(getName(), partCols,
new ObjectPath(hiveTable.getDbName(), hiveTable.getTableName()), partitionSpec);
}
@@ -1116,7 +1124,7 @@ public class HiveCatalog extends AbstractCatalog {
@Override
public List<String> listFunctions(String databaseName) throws DatabaseNotExistException, CatalogException {
- checkArgument(!StringUtils.isNullOrWhitespaceOnly(databaseName), "databaseName cannot be null or empty");
+ checkArgument(!isNullOrWhitespaceOnly(databaseName), "databaseName cannot be null or empty");
// client.getFunctions() returns empty list when the database doesn't exist
// thus we need to explicitly check whether the database exists or not
@@ -1472,6 +1480,6 @@ public class HiveCatalog extends AbstractCatalog {
@VisibleForTesting
public static boolean isEmbeddedMetastore(HiveConf hiveConf) {
- return StringUtils.isNullOrWhitespaceOnly(hiveConf.getVar(HiveConf.ConfVars.METASTOREURIS));
+ return isNullOrWhitespaceOnly(hiveConf.getVar(HiveConf.ConfVars.METASTOREURIS));
}
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/descriptors/HiveCatalogValidator.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/descriptors/HiveCatalogValidator.java
index 6218829..40d5dd0 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/descriptors/HiveCatalogValidator.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/descriptors/HiveCatalogValidator.java
@@ -28,6 +28,7 @@ public class HiveCatalogValidator extends CatalogDescriptorValidator {
public static final String CATALOG_TYPE_VALUE_HIVE = "hive";
public static final String CATALOG_HIVE_CONF_DIR = "hive-conf-dir";
public static final String CATALOG_HIVE_VERSION = "hive-version";
+ public static final String CATALOG_HADOOP_CONF_DIR = "hadoop-conf-dir";
@Override
public void validate(DescriptorProperties properties) {
@@ -35,5 +36,6 @@ public class HiveCatalogValidator extends CatalogDescriptorValidator {
properties.validateValue(CATALOG_TYPE, CATALOG_TYPE_VALUE_HIVE, false);
properties.validateString(CATALOG_HIVE_CONF_DIR, true, 1);
properties.validateString(CATALOG_HIVE_VERSION, true, 1);
+ properties.validateString(CATALOG_HADOOP_CONF_DIR, true, 1);
}
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/factories/HiveCatalogFactory.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/factories/HiveCatalogFactory.java
index 5b39ef8..c4e8127 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/factories/HiveCatalogFactory.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/factories/HiveCatalogFactory.java
@@ -34,6 +34,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
+import static org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator.CATALOG_HADOOP_CONF_DIR;
import static org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator.CATALOG_HIVE_CONF_DIR;
import static org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator.CATALOG_HIVE_VERSION;
import static org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator.CATALOG_TYPE_VALUE_HIVE;
@@ -66,6 +67,8 @@ public class HiveCatalogFactory implements CatalogFactory {
properties.add(CATALOG_HIVE_VERSION);
+ properties.add(CATALOG_HADOOP_CONF_DIR);
+
return properties;
}
@@ -79,9 +82,11 @@ public class HiveCatalogFactory implements CatalogFactory {
final Optional<String> hiveConfDir = descriptorProperties.getOptionalString(CATALOG_HIVE_CONF_DIR);
+ final Optional<String> hadoopConfDir = descriptorProperties.getOptionalString(CATALOG_HADOOP_CONF_DIR);
+
final String version = descriptorProperties.getOptionalString(CATALOG_HIVE_VERSION).orElse(HiveShimLoader.getHiveVersion());
- return new HiveCatalog(name, defaultDatabase, hiveConfDir.orElse(null), version);
+ return new HiveCatalog(name, defaultDatabase, hiveConfDir.orElse(null), hadoopConfDir.orElse(null), version);
}
private static DescriptorProperties getValidatedProperties(Map<String, String> properties) {
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTableUtil.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTableUtil.java
index db36ceb..4b0e6a1 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTableUtil.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTableUtil.java
@@ -50,6 +50,8 @@ import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalTypeFamily;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.TableType;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
@@ -62,6 +64,7 @@ import org.apache.hadoop.hive.ql.io.StorageFormatFactory;
import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
+import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -421,6 +424,37 @@ public class HiveTableUtil {
}
}
+ /**
+ * Returns a new Hadoop Configuration object using the path to the hadoop conf configured.
+ *
+ * @param hadoopConfDir Hadoop conf directory path.
+ * @return A Hadoop configuration instance.
+ */
+ public static Configuration getHadoopConfiguration(String hadoopConfDir) {
+ if (new File(hadoopConfDir).exists()) {
+ Configuration hadoopConfiguration = new Configuration();
+ File coreSite = new File(hadoopConfDir, "core-site.xml");
+ if (coreSite.exists()) {
+ hadoopConfiguration.addResource(new Path(coreSite.getAbsolutePath()));
+ }
+ File hdfsSite = new File(hadoopConfDir, "hdfs-site.xml");
+ if (hdfsSite.exists()) {
+ hadoopConfiguration.addResource(new Path(hdfsSite.getAbsolutePath()));
+ }
+ File yarnSite = new File(hadoopConfDir, "yarn-site.xml");
+ if (yarnSite.exists()) {
+ hadoopConfiguration.addResource(new Path(yarnSite.getAbsolutePath()));
+ }
+ // Add mapred-site.xml. We need to read configurations like compression codec.
+ File mapredSite = new File(hadoopConfDir, "mapred-site.xml");
+ if (mapredSite.exists()) {
+ hadoopConfiguration.addResource(new Path(mapredSite.getAbsolutePath()));
+ }
+ return hadoopConfiguration;
+ }
+ return null;
+ }
+
private static class ExpressionExtractor implements ExpressionVisitor<String> {
// maps a supported function to its name
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveTestUtils.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveTestUtils.java
index a30cf14..8aa25b8 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveTestUtils.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveTestUtils.java
@@ -72,6 +72,11 @@ public class HiveTestUtils {
StringUtils.isNullOrWhitespaceOnly(hiveVersion) ? HiveShimLoader.getHiveVersion() : hiveVersion, true);
}
+ public static HiveCatalog createHiveCatalog(String name, String hiveConfDir, String hadoopConfDir, String hiveVersion) {
+ return new HiveCatalog(name, null, hiveConfDir, hadoopConfDir,
+ StringUtils.isNullOrWhitespaceOnly(hiveVersion) ? HiveShimLoader.getHiveVersion() : hiveVersion);
+ }
+
public static HiveCatalog createHiveCatalog(HiveConf hiveConf) {
return new HiveCatalog(CatalogTest.TEST_CATALOG_NAME, null, hiveConf, HiveShimLoader.getHiveVersion(), true);
}
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/factories/HiveCatalogFactoryTest.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/factories/HiveCatalogFactoryTest.java
index 6a838bd..387f4ab 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/factories/HiveCatalogFactoryTest.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/table/catalog/hive/factories/HiveCatalogFactoryTest.java
@@ -41,6 +41,7 @@ import java.net.URL;
import java.util.HashMap;
import java.util.Map;
+import static org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator.CATALOG_HADOOP_CONF_DIR;
import static org.junit.Assert.assertEquals;
/**
@@ -57,7 +58,7 @@ public class HiveCatalogFactoryTest extends TestLogger {
public ExpectedException expectedException = ExpectedException.none();
@Test
- public void test() {
+ public void testCreateHiveCatalog() {
final String catalogName = "mycatalog";
final HiveCatalog expectedCatalog = HiveTestUtils.createHiveCatalog(catalogName, null);
@@ -74,6 +75,31 @@ public class HiveCatalogFactoryTest extends TestLogger {
}
@Test
+ public void testCreateHiveCatalogWithHadoopConfDir() throws IOException {
+ final String catalogName = "mycatalog";
+
+ final String hadoopConfDir = tempFolder.newFolder().getAbsolutePath();
+ final File mapredSiteFile = new File(hadoopConfDir, "mapred-site.xml");
+ final String mapredKey = "mapred.site.config.key";
+ final String mapredVal = "mapred.site.config.val";
+ writeProperty(mapredSiteFile, mapredKey, mapredVal);
+
+ final HiveCatalog expectedCatalog = HiveTestUtils.createHiveCatalog(catalogName, CONF_DIR.getPath(), hadoopConfDir, null);
+
+ final HiveCatalogDescriptor catalogDescriptor = new HiveCatalogDescriptor();
+ catalogDescriptor.hiveSitePath(CONF_DIR.getPath());
+
+ final Map<String, String> properties = new HashMap<>(catalogDescriptor.toProperties());
+ properties.put(CATALOG_HADOOP_CONF_DIR, hadoopConfDir);
+
+ final Catalog actualCatalog = TableFactoryService.find(CatalogFactory.class, properties)
+ .createCatalog(catalogName, properties);
+
+ checkEquals(expectedCatalog, (HiveCatalog) actualCatalog);
+ assertEquals(mapredVal, ((HiveCatalog) actualCatalog).getHiveConf().get(mapredKey));
+ }
+
+ @Test
public void testLoadHadoopConfigFromEnv() throws IOException {
Map<String, String> customProps = new HashMap<>();
String k1 = "what is connector?";