This is an automated email from the ASF dual-hosted git repository.
wenjun pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 66ec2807 [Improvement][hudi-spark-connector] Refactored hudi config
parameters (#1735)
66ec2807 is described below
commit 66ec2807b2b609d9039601b23732ba6f8d1675f5
Author: mans2singh <[email protected]>
AuthorDate: Mon Apr 25 22:11:42 2022 -0400
[Improvement][hudi-spark-connector] Refactored hudi config parameters
(#1735)
Refactored hudi config parameters
---
.../org/apache/seatunnel/spark/hudi/Config.scala | 48 ++++++++++++++++++++++
.../apache/seatunnel/spark/hudi/sink/Hudi.scala | 9 ++--
.../apache/seatunnel/spark/hudi/source/Hudi.scala | 5 ++-
3 files changed, 56 insertions(+), 6 deletions(-)
diff --git
a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hudi/src/main/scala/org/apache/seatunnel/spark/hudi/Config.scala
b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hudi/src/main/scala/org/apache/seatunnel/spark/hudi/Config.scala
new file mode 100644
index 00000000..5fe392e8
--- /dev/null
+++
b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hudi/src/main/scala/org/apache/seatunnel/spark/hudi/Config.scala
@@ -0,0 +1,48 @@
+/*
+ * 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.seatunnel.spark.hudi
+
+/**
+ * Configurations and defaults for Hudi source and sink
+ */
+object Config extends Serializable {
+
+ /**
+ * Hoodie base path
+ */
+ val HOODIE_BASE_PATH = "hoodie.base.path"
+
+ /**
+ * Hoodie table name
+ */
+ val HOODIE_TABLE_NAME = "hoodie.table.name"
+
+ /**
+ * Save mode
+ */
+ val SAVE_MODE = "save_mode"
+
+ /**
+ * Default save mode
+ */
+ val DEFAULT_SAVE_MODE = "append"
+
+ /**
+ * Hoodie data store read paths
+ */
+ val HOODIE_DATASTORE_READ_PATHS = "hoodie.datasource.read.paths"
+}
diff --git
a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hudi/src/main/scala/org/apache/seatunnel/spark/hudi/sink/Hudi.scala
b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hudi/src/main/scala/org/apache/seatunnel/spark/hudi/sink/Hudi.scala
index 32cb6cf2..ac335213 100644
---
a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hudi/src/main/scala/org/apache/seatunnel/spark/hudi/sink/Hudi.scala
+++
b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hudi/src/main/scala/org/apache/seatunnel/spark/hudi/sink/Hudi.scala
@@ -19,6 +19,7 @@ package org.apache.seatunnel.spark.hudi.sink
import org.apache.seatunnel.common.config.CheckConfigUtil.checkAllExists
import org.apache.seatunnel.common.config.CheckResult
import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory
+import org.apache.seatunnel.spark.hudi.Config.{DEFAULT_SAVE_MODE,
HOODIE_BASE_PATH, HOODIE_TABLE_NAME, SAVE_MODE}
import org.apache.seatunnel.spark.SparkEnvironment
import org.apache.seatunnel.spark.batch.SparkBatchSink
import org.apache.spark.sql.{Dataset, Row}
@@ -28,13 +29,13 @@ import scala.collection.JavaConversions._
class Hudi extends SparkBatchSink {
override def checkConfig(): CheckResult = {
- checkAllExists(config, "hoodie.base.path", "hoodie.table.name")
+ checkAllExists(config, HOODIE_BASE_PATH, HOODIE_TABLE_NAME)
}
override def prepare(env: SparkEnvironment): Unit = {
val defaultConfig = ConfigFactory.parseMap(
Map(
- "save_mode" -> "append"))
+ SAVE_MODE -> DEFAULT_SAVE_MODE))
config = config.withFallback(defaultConfig)
}
@@ -43,8 +44,8 @@ class Hudi extends SparkBatchSink {
for (e <- config.entrySet()) {
writer.option(e.getKey, String.valueOf(e.getValue.unwrapped()))
}
- writer.mode(config.getString("save_mode"))
- .save(config.getString("hoodie.base.path"))
+ writer.mode(config.getString(SAVE_MODE))
+ .save(config.getString(HOODIE_BASE_PATH))
}
override def getPluginName: String = "Hudi"
diff --git
a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hudi/src/main/scala/org/apache/seatunnel/spark/hudi/source/Hudi.scala
b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hudi/src/main/scala/org/apache/seatunnel/spark/hudi/source/Hudi.scala
index 723077f8..82e55f09 100644
---
a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hudi/src/main/scala/org/apache/seatunnel/spark/hudi/source/Hudi.scala
+++
b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-hudi/src/main/scala/org/apache/seatunnel/spark/hudi/source/Hudi.scala
@@ -18,6 +18,7 @@ package org.apache.seatunnel.spark.hudi.source
import org.apache.seatunnel.common.config.CheckConfigUtil.checkAllExists
import org.apache.seatunnel.common.config.CheckResult
+import org.apache.seatunnel.spark.hudi.Config.HOODIE_DATASTORE_READ_PATHS
import org.apache.seatunnel.spark.SparkEnvironment
import org.apache.seatunnel.spark.batch.SparkBatchSource
import org.apache.spark.sql.{Dataset, Row}
@@ -27,7 +28,7 @@ import scala.collection.JavaConversions._
class Hudi extends SparkBatchSource {
override def checkConfig(): CheckResult = {
- checkAllExists(config, "hoodie.datasource.read.paths")
+ checkAllExists(config, HOODIE_DATASTORE_READ_PATHS)
}
override def getData(env: SparkEnvironment): Dataset[Row] = {
@@ -37,7 +38,7 @@ class Hudi extends SparkBatchSource {
reader.option(e.getKey, String.valueOf(e.getValue.unwrapped()))
}
- reader.load(config.getString("hoodie.datasource.read.paths"))
+ reader.load(config.getString(HOODIE_DATASTORE_READ_PATHS))
}
override def getPluginName: String = "Hudi"
