refactor SnowflakeInput and SnowflakeOutput to include required name field; update configuration examples to use lowercase keys

Author: marclamy

build.sbt

@@ -179,4 +179,4 @@ lazy val root = (project in file("."))
     name := "dataio",
     publish / skip := true
   )
-  .aggregate(core, test, kafka)
+  .aggregate(core, test, kafka, snowflake)

docs/content/_data/config/pipes/snowflake/batch.yaml

@@ -5,8 +5,8 @@ input:
 output:
   type: com.amadeus.dataio.pipes.snowflake.batch.SnowflakeOutput
   fields:
-    - name: Mode
+    - name: mode
       mandatory: Yes
       description: Writing mode on the Snowflake table
-      example: Mode = "append"
+      example: mode = "append"
 

docs/content/_data/config/pipes/snowflake/common.yaml

@@ -8,9 +8,6 @@ links:
     url: https://docs.snowflake.com/en/user-guide/spark-connector-use#using-the-connector-in-scala
 
 fields:
-  - name: Options
+  - name: options
     description: Snowflake options to specify such as key = value pairs. These options are then passed as option to the Spark connector for Snowflake
-    example: Options { Database = "example_database"
-      Schema = "example_schema"
-      ...
-      }
+    example: options {"sfDatabase" = "my_db" }

docs/content/_data/config/pipes/snowflake/streaming.yaml

@@ -0,0 +1,5 @@
+package com.amadeus.dataio.pipes.snowflake
+
+object SnowflakeCommons {
+  val SNOWFLAKE_CONNECTOR_NAME = "net.snowflake.spark.snowflake"
+}

snowflake/src/main/scala/com/amadeus/dataio/pipes/snowflake/SnowflakeCommons.scala

@@ -2,46 +2,46 @@ package com.amadeus.dataio.pipes.snowflake.batch
 
 import com.amadeus.dataio.core.{Input, Logging}
 import com.amadeus.dataio.config.fields._
+import com.amadeus.dataio.pipes.snowflake.SnowflakeCommons.SNOWFLAKE_CONNECTOR_NAME
 import com.typesafe.config.{Config, ConfigFactory}
 import org.apache.spark.sql.{DataFrame, SparkSession}
 
-/**
- * Class for reading Snowflake input
+import scala.util.Try
+
+/** Class for reading Snowflake input.
  *
- * @param options the snowflake connector options.
  * @param config Contains the Typesafe Config object that was used at instantiation to configure this entity.
  */
 case class SnowflakeInput(
-    options: Map[String, String],
+                           name: String,
+                           options: Map[String, String] = Map(),
     config: Config = ConfigFactory.empty()
 ) extends Input
     with Logging {
-
-  val SNOWFLAKE_CONNECTOR_NAME = "net.snowflake.spark.snowflake"
-
-  /**
-   * Reads a batch of data from snowflake.
-   *
-   * @param spark The SparkSession which will be used to read the data.
-   * @return The data that was read.
-   * @throws Exception If the exactly one of the dateRange/dateColumn fields is None.
-   */
   override def read(implicit spark: SparkSession): DataFrame = {
+    logger.info(s"reading: $name")
+    if (options.nonEmpty) logger.info(s"options: $options")
+
     spark.read.format(SNOWFLAKE_CONNECTOR_NAME).options(options).load()
   }
 
 }
 
 object SnowflakeInput {
 
-  /**
-   * Creates a new instance of SnowflakeInput from a typesafe Config object.
+  /** Creates a new instance of SnowflakeInput from a typesafe Config object.
    *
    * @param config typesafe Config object containing the configuration fields.
    * @return a new SnowflakeInput object.
    * @throws com.typesafe.config.ConfigException If any of the mandatory fields is not available in the config argument.
    */
   def apply(implicit config: Config): SnowflakeInput = {
-    SnowflakeInput(options = getOptions, config = config)
+    val name = Try {
+      config.getString("name")
+    } getOrElse {
+      throw new Exception("Missing required `name` field in configuration.")
+    }
+
+    SnowflakeInput(name, options = getOptions, config = config)
   }
 }

snowflake/src/main/scala/com/amadeus/dataio/pipes/snowflake/batch/SnowflakeInput.scala

@@ -2,32 +2,28 @@ package com.amadeus.dataio.pipes.snowflake.batch
 
 import com.amadeus.dataio.config.fields._
 import com.amadeus.dataio.core.{Logging, Output}
+import com.amadeus.dataio.pipes.snowflake.SnowflakeCommons.SNOWFLAKE_CONNECTOR_NAME
 import com.typesafe.config.{Config, ConfigFactory}
 import org.apache.spark.sql.{Dataset, SparkSession}
 
-/**
- * Allows to write data to Snowflake.
+import scala.util.Try
+
+/** Allows to write data to Snowflake.
  *
- * @param mode the mode to use.
- * @param options the snowflake connector options.
  * @param config the config object.
  */
 case class SnowflakeOutput(
+                            name: String,
     mode: String,
-    options: Map[String, String],
+                            options: Map[String, String] = Map(),
     config: Config = ConfigFactory.empty()
 ) extends Output
     with Logging {
 
-  val SNOWFLAKE_CONNECTOR_NAME = "net.snowflake.spark.snowflake"
-
-  /**
-   * Writes data to this output.
-   *
-   * @param data  The data to write.
-   * @param spark The SparkSession which will be used to write the data.
-   */
   override def write[T](data: Dataset[T])(implicit spark: SparkSession): Unit = {
+    logger.info(s"reading: $name")
+    if (options.nonEmpty) logger.info(s"options: $options")
+    logger.info(s"mode: $mode")
 
     data.write
       .format(SNOWFLAKE_CONNECTOR_NAME)
@@ -39,18 +35,26 @@ case class SnowflakeOutput(
 
 object SnowflakeOutput {
 
-  /**
-   * Creates a new instance of SnowflakeOutput from a typesafe Config object.
+  /** Creates a new instance of SnowflakeOutput from a typesafe Config object.
    *
    * @param config typesafe Config object containing the configuration fields.
    * @return a new SnowflakeOutput object.
    * @throws com.typesafe.config.ConfigException If any of the mandatory fields is not available in the config argument.
    */
   def apply(implicit config: Config): SnowflakeOutput = {
-
-    val mode = config.getString("Mode")
-
-    SnowflakeOutput(mode = mode, options = getOptions, config = config)
+    val name = Try {
+      config.getString("name")
+    } getOrElse {
+      throw new Exception("Missing required `name` field in configuration.")
+    }
+
+    val mode = Try {
+      config.getString("mode")
+    } getOrElse {
+      throw new Exception("Missing required `mode` field in configuration.")
+    }
+
+    SnowflakeOutput(name, mode = mode, options = getOptions, config = config)
   }
 
 }