This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 7554ea4d3 [INLONG-7655][Manager] Remove hudi-flink1.13-bundle
dependency in Manager-service (#7657)
7554ea4d3 is described below
commit 7554ea4d3a90ccfbf3002c27491ccb698b7934c6
Author: feat <[email protected]>
AuthorDate: Tue Mar 21 16:58:03 2023 +0800
[INLONG-7655][Manager] Remove hudi-flink1.13-bundle dependency in
Manager-service (#7657)
---
inlong-manager/manager-service/pom.xml | 4 -
.../resource/sink/hudi/HudiCatalogClient.java | 29 +++---
.../service/resource/sink/hudi/HudiFileFormat.java | 30 ++++++
.../service/resource/sink/hudi/HudiUtils.java | 116 +++++++++++++++++++++
4 files changed, 160 insertions(+), 19 deletions(-)
diff --git a/inlong-manager/manager-service/pom.xml
b/inlong-manager/manager-service/pom.xml
index 88c40bc77..194152175 100644
--- a/inlong-manager/manager-service/pom.xml
+++ b/inlong-manager/manager-service/pom.xml
@@ -557,10 +557,6 @@
<artifactId>flink-core</artifactId>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.apache.hudi</groupId>
- <artifactId>hudi-flink1.13-bundle</artifactId>
- </dependency>
<dependency>
<groupId>org.apache.kudu</groupId>
<artifactId>kudu-client</artifactId>
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/hudi/HudiCatalogClient.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/hudi/HudiCatalogClient.java
index cff7995f7..72eee8360 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/hudi/HudiCatalogClient.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/hudi/HudiCatalogClient.java
@@ -17,6 +17,8 @@
package org.apache.inlong.manager.service.resource.sink.hudi;
+import com.google.common.collect.Maps;
+
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
@@ -34,18 +36,17 @@ import
org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hudi.avro.HoodieAvroUtils;
-import org.apache.hudi.common.model.HoodieFileFormat;
-import org.apache.hudi.exception.HoodieCatalogException;
-import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils;
-import
org.apache.hudi.org.apache.hbase.thirdparty.com.google.common.collect.Maps;
-import org.apache.hudi.sync.common.util.ConfigUtils;
import org.apache.inlong.manager.pojo.sink.hudi.HudiColumnInfo;
import org.apache.inlong.manager.pojo.sink.hudi.HudiTableInfo;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static
org.apache.inlong.manager.service.resource.sink.hudi.HudiUtils.IS_QUERY_AS_RO_TABLE;
+import static
org.apache.inlong.manager.service.resource.sink.hudi.HudiUtils.getInputFormatClassName;
+import static
org.apache.inlong.manager.service.resource.sink.hudi.HudiUtils.getOutputFormatClassName;
+import static
org.apache.inlong.manager.service.resource.sink.hudi.HudiUtils.getSerDeClassName;
+
/**
* The Catalog client for Hudi.
*/
@@ -53,14 +54,12 @@ public class HudiCatalogClient {
private static final Logger LOG =
LoggerFactory.getLogger(HudiCatalogClient.class);
- private final String uri;
private final String dbName;
private final String warehouse;
private IMetaStoreClient client;
private final HiveConf hiveConf;
public HudiCatalogClient(String uri, String warehouse, String dbName)
throws MetaException {
- this.uri = uri;
this.warehouse = warehouse;
this.dbName = dbName;
hiveConf = new HiveConf();
@@ -76,7 +75,7 @@ public class HudiCatalogClient {
try {
this.client = Hive.get(hiveConf).getMSC();
} catch (Exception e) {
- throw new HoodieCatalogException("Failed to create hive
metastore client", e);
+ throw new RuntimeException("Failed to create hive metastore
client", e);
}
LOG.info("Connected to Hive metastore");
}
@@ -139,7 +138,7 @@ public class HudiCatalogClient {
Table hiveTable = client.getTable(dbName, tableName);
List<FieldSchema> allCols = hiveTable.getSd().getCols().stream()
// filter out the metadata columns
- .filter(s -> !HoodieAvroUtils.isMetadataField(s.getName()))
+ .filter(s -> !HudiUtils.isMetadataFile(s.getName()))
.collect(Collectors.toList());
return allCols.stream()
@@ -207,18 +206,18 @@ public class HudiCatalogClient {
hiveTable.setTableName(tableName);
// FIXME: splitSchemas need config by frontend
- HoodieFileFormat baseFileFormat = HoodieFileFormat.PARQUET;
+ HudiFileFormat baseFileFormat = HudiFileFormat.PARQUET;
// ignore uber input Format
String inputFormatClassName =
- HoodieInputFormatUtils.getInputFormatClassName(baseFileFormat,
useRealTimeInputFormat);
- String outputFormatClassName =
HoodieInputFormatUtils.getOutputFormatClassName(baseFileFormat);
- String serDeClassName =
HoodieInputFormatUtils.getSerDeClassName(baseFileFormat);
+ getInputFormatClassName(baseFileFormat,
useRealTimeInputFormat);
+ String outputFormatClassName =
getOutputFormatClassName(baseFileFormat);
+ String serDeClassName = getSerDeClassName(baseFileFormat);
sd.setInputFormat(inputFormatClassName);
sd.setOutputFormat(outputFormatClassName);
Map<String, String> serdeProperties = new HashMap<>();
serdeProperties.put("path", location);
- serdeProperties.put(ConfigUtils.IS_QUERY_AS_RO_TABLE,
String.valueOf(!useRealTimeInputFormat));
+ serdeProperties.put(IS_QUERY_AS_RO_TABLE,
String.valueOf(!useRealTimeInputFormat));
sd.setSerdeInfo(new SerDeInfo(null, serDeClassName, serdeProperties));
sd.setLocation(location);
hiveTable.setSd(sd);
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/hudi/HudiFileFormat.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/hudi/HudiFileFormat.java
new file mode 100644
index 000000000..8429b53ba
--- /dev/null
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/hudi/HudiFileFormat.java
@@ -0,0 +1,30 @@
+/*
+ * 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 org.apache.inlong.manager.service.resource.sink.hudi;
+
+/**
+ * Hudi file format.
+ */
+public enum HudiFileFormat {
+
+ PARQUET,
+ HUDI_LOG,
+ HFILE,
+ ORC;
+
+}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/hudi/HudiUtils.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/hudi/HudiUtils.java
new file mode 100644
index 000000000..577952713
--- /dev/null
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/hudi/HudiUtils.java
@@ -0,0 +1,116 @@
+/*
+ * 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 org.apache.inlong.manager.service.resource.sink.hudi;
+
+import com.google.common.collect.Sets;
+
+import java.util.Set;
+
+/**
+ * The utility of hudi.
+ */
+public class HudiUtils {
+
+ private static final String COMMIT_TIME_METADATA_FILE_NAME =
"_Hudi_commit_time";
+ private static final String COMMIT_SEQNO_METADATA_FILE_NAME =
"_Hudi_commit_seqno";
+ private static final String RECORD_KEY_METADATA_FILE_NAME =
"_Hudi_record_key";
+ private static final String PARTITION_PATH_METADATA_FILE_NAME =
"_Hudi_partition_path";
+ private static final String METADATA_FILE_NAME = "_Hudi_file_name";
+ private static final String OPERATION_METADATA_FILE_NAME =
"_Hudi_operation";
+
+ public static final String IS_QUERY_AS_RO_TABLE = "Hudi.query.as.ro.table";
+
+ private static final Set<String> HUDI_METADATA_FILES =
+ Sets.newHashSet(COMMIT_TIME_METADATA_FILE_NAME,
COMMIT_SEQNO_METADATA_FILE_NAME,
+ RECORD_KEY_METADATA_FILE_NAME,
PARTITION_PATH_METADATA_FILE_NAME, METADATA_FILE_NAME,
+ OPERATION_METADATA_FILE_NAME);
+ private static final String PARQUET_REALTIME_INPUT_FORMAT_NAME =
+ "org.apache.hudi.hadoop.realtime.HudiParquetRealtimeInputFormat";
+ private static final String PARQUET_INPUT_FORMAT_NAME =
"org.apache.hudi.hadoop.HudiParquetInputFormat";
+ private static final String HFILE_REALTIME_INPUT_FORMAT_NAME =
+ "org.apache.hudi.hadoop.realtime.HudiHFileRealtimeInputFormat";
+ private static final String HFILE_INPUT_FORMAT_NAME =
"org.apache.hudi.hadoop.HudiHFileInputFormat";
+ private static final String ORC_INPUT_FORMAT_NAME =
"org.apache.hadoop.hive.ql.io.orc.OrcInputFormat";
+ private static final String PARQUET_HIVE_SER_DE_CLASS_NAME =
+ "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe";
+ private static final String ORC_SER_DE_CLASS_NAME =
"org.apache.hadoop.hive.ql.io.orc.OrcSerde";
+ private static final String MAPRED_PARQUET_OUTPUT_FORMAT_NAME =
+ "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat";
+ private static final String ORC_OUTPUT_FORMAT_NAME =
"org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat";
+
+ /**
+ * Check if metadata file.
+ */
+ public static boolean isMetadataFile(String fileName) {
+ return HUDI_METADATA_FILES.contains(fileName);
+ }
+
+ /**
+ * Get the InputFormat class name.
+ */
+ public static String getInputFormatClassName(HudiFileFormat
baseFileFormat, boolean realtime) {
+ switch (baseFileFormat) {
+ case PARQUET:
+ if (realtime) {
+ return PARQUET_REALTIME_INPUT_FORMAT_NAME;
+ } else {
+ return PARQUET_INPUT_FORMAT_NAME;
+ }
+ case HFILE:
+ if (realtime) {
+ return HFILE_REALTIME_INPUT_FORMAT_NAME;
+ } else {
+ return HFILE_INPUT_FORMAT_NAME;
+ }
+ case ORC:
+ return ORC_INPUT_FORMAT_NAME;
+ default:
+ throw new RuntimeException("Hudi InputFormat not implemented
for base file format " + baseFileFormat);
+ }
+ }
+
+ /**
+ * Get the OutputFormat class name.
+ */
+ public static String getOutputFormatClassName(HudiFileFormat
baseFileFormat) {
+ switch (baseFileFormat) {
+ case PARQUET:
+ case HFILE:
+ return MAPRED_PARQUET_OUTPUT_FORMAT_NAME;
+ case ORC:
+ return ORC_OUTPUT_FORMAT_NAME;
+ default:
+ throw new RuntimeException("No OutputFormat for base file
format " + baseFileFormat);
+ }
+ }
+
+ /**
+ * Get the Ser and DeSer class name.
+ */
+ public static String getSerDeClassName(HudiFileFormat baseFileFormat) {
+ switch (baseFileFormat) {
+ case PARQUET:
+ case HFILE:
+ return PARQUET_HIVE_SER_DE_CLASS_NAME;
+ case ORC:
+ return ORC_SER_DE_CLASS_NAME;
+ default:
+ throw new RuntimeException("No SerDe for base file format " +
baseFileFormat);
+ }
+ }
+}
