[incubator-seatunnel] branch dev updated: [Improvement][connector-spark-iceberg] Refactored config arguments and defaults (#1782)

[we...@apache.org]

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 6b42a6b5 [Improvement][connector-spark-iceberg] Refactored config arguments and defaults (#1782)
6b42a6b5 is described below

commit 6b42a6b503986425410331bb9c0cf5f3a9ca9671
Author: mans2singh <ma...@users.noreply.github.com>
AuthorDate: Tue May 3 04:47:57 2022 -0400

    [Improvement][connector-spark-iceberg] Refactored config arguments and defaults (#1782)
    
    * Refactored config arguments and defaults
---
 .../apache/seatunnel/spark/iceberg/Config.scala    | 43 ++++++++++++++++++++++
 .../seatunnel/spark/iceberg/sink/Iceberg.scala     |  9 +++--
 .../seatunnel/spark/iceberg/source/Iceberg.scala   | 11 ++++--
 3 files changed, 55 insertions(+), 8 deletions(-)

diff --git a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-iceberg/src/main/scala/org/apache/seatunnel/spark/iceberg/Config.scala b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-iceberg/src/main/scala/org/apache/seatunnel/spark/iceberg/Config.scala
new file mode 100644
index 00000000..e316a6c0
--- /dev/null
+++ b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-iceberg/src/main/scala/org/apache/seatunnel/spark/iceberg/Config.scala
@@ -0,0 +1,43 @@
+/*
+ * 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.iceberg
+
+/**
+ * Configurations and defaults for Iceberg source and sink
+ */
+object Config extends Serializable {
+
+  /**
+   * Save mode config
+   */
+  val SAVE_MODE = "saveMode"
+
+  /**
+   * Default save mode value
+   */
+  val SAVE_MODE_DEFAULT = "append"
+
+  /**
+   * Path config
+   */
+  val PATH = "path"
+
+  /**
+   * Pre sql config
+   */
+  val PRE_SQL = "pre_sql"
+}
diff --git a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-iceberg/src/main/scala/org/apache/seatunnel/spark/iceberg/sink/Iceberg.scala b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-iceberg/src/main/scala/org/apache/seatunnel/spark/iceberg/sink/Iceberg.scala
index ee42b36e..2f15e236 100644
--- a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-iceberg/src/main/scala/org/apache/seatunnel/spark/iceberg/sink/Iceberg.scala
+++ b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-iceberg/src/main/scala/org/apache/seatunnel/spark/iceberg/sink/Iceberg.scala
@@ -21,6 +21,7 @@ import org.apache.seatunnel.common.config.CheckResult
 import org.apache.seatunnel.shade.com.typesafe.config.{ConfigFactory, ConfigValueType}
 import org.apache.seatunnel.spark.SparkEnvironment
 import org.apache.seatunnel.spark.batch.SparkBatchSink
+import org.apache.seatunnel.spark.iceberg.Config.{PATH, SAVE_MODE, SAVE_MODE_DEFAULT}
 import org.apache.spark.sql.{Dataset, Row}
 
 import scala.collection.JavaConversions._
@@ -39,18 +40,18 @@ class Iceberg extends SparkBatchSink {
           writer.option(e.getKey, config.getString(e.getKey))
       }
     }
-    writer.mode(config.getString("saveMode"))
-      .save(config.getString("path"))
+    writer.mode(config.getString(SAVE_MODE))
+      .save(config.getString(PATH))
   }
 
   override def checkConfig(): CheckResult = {
-    checkAllExists(config, "path")
+    checkAllExists(config, PATH)
   }
 
   override def prepare(prepareEnv: SparkEnvironment): Unit = {
     val defaultConfig = ConfigFactory.parseMap(
       Map(
-        "saveMode" -> "append"))
+        SAVE_MODE -> SAVE_MODE_DEFAULT))
     config = config.withFallback(defaultConfig)
   }
 
diff --git a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-iceberg/src/main/scala/org/apache/seatunnel/spark/iceberg/source/Iceberg.scala b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-iceberg/src/main/scala/org/apache/seatunnel/spark/iceberg/source/Iceberg.scala
index 2bed3f11..ade58126 100644
--- a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-iceberg/src/main/scala/org/apache/seatunnel/spark/iceberg/source/Iceberg.scala
+++ b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-iceberg/src/main/scala/org/apache/seatunnel/spark/iceberg/source/Iceberg.scala
@@ -16,16 +16,19 @@
  */
 package org.apache.seatunnel.spark.iceberg.source
 
+import org.apache.seatunnel.apis.base.plugin.Plugin
 import org.apache.seatunnel.common.config.CheckConfigUtil.checkAllExists
 import org.apache.seatunnel.common.config.CheckResult
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigValueType
 import org.apache.seatunnel.spark.SparkEnvironment
 import org.apache.seatunnel.spark.batch.SparkBatchSource
+import org.apache.seatunnel.spark.iceberg.Config.{PATH, PRE_SQL}
 import org.apache.spark.sql.{Dataset, Row}
 
 import scala.collection.JavaConversions._
 
 class Iceberg extends SparkBatchSource {
+
   override def getData(env: SparkEnvironment): Dataset[Row] = {
     val reader = env.getSparkSession.read.format("iceberg")
     for (e <- config.entrySet()) {
@@ -38,13 +41,13 @@ class Iceberg extends SparkBatchSource {
           reader.option(e.getKey, config.getString(e.getKey))
       }
     }
-    val df = reader.load(config.getString("path"))
-    df.createOrReplaceTempView(config.getString("result_table_name"))
-    env.getSparkSession.sql(config.getString("pre_sql"))
+    val df = reader.load(config.getString(PATH))
+    df.createOrReplaceTempView(config.getString(Plugin.RESULT_TABLE_NAME))
+    env.getSparkSession.sql(config.getString(PRE_SQL))
   }
 
   override def checkConfig(): CheckResult = {
-    checkAllExists(config, "path", "pre_sql")
+    checkAllExists(config, PATH, PRE_SQL)
   }
 
   override def getPluginName: String = "Iceberg"