feat: add ability to use spark reader native implemention (#193)

* feat: add ability to use spark reader native implemention * fix: typo in FileConnector * feat: use file connector config for native option * feat: add unit tests for native * fix: deprecated method in FileConnector unit tests * feat: replace native option by pathFormat enum option * fix: typo in exception message for pathFormat * feat: add unit tests for unsupported path format option * Update configuration_example.md Co-authored-by: Marouane Felja <marouane.felja@idemia.com> Co-authored-by: Qinx <17144939+qxzzxq@users.noreply.github.com>

Author: 3 people

docs/data_access_layer/configuration_example.md

@@ -19,6 +19,7 @@
 | emptyValue | sets the string representation of an empty value | true, default empty string| |
 | dateFormat | sets the string that indicates a date format | true, default `yyyy-MM-dd`| |
 | timestampFormat | sets the string that indicates a timestamp format | true, default `yyyy-MM-dd'T'HH:mm:ss.SSSXXX`| |
+| pathFormat | choose between REGEX path format (ex: `^s3:\/\/bucket\/\/col1=B\/col2=([A-C])$`) or WILDCARD path format (ex: `s3://bucket/*/*A*/internal-*.csv`) | true, default `REGEX`| `WILDCARD` |
 
 For other options, please refer to [this doc](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/DataFrameReader.html).
 
@@ -69,6 +70,7 @@ csvWithSchema {
 | dateFormat | sets the string that indicates a date format | true, default `yyyy-MM-dd`| |
 | timestampFormat | sets the string that indicates a timestamp format | true, default `yyyy-MM-dd'T'HH:mm:ss.SSSXXX`| |
 | dropFieldIfAllNull | whether to ignore column of all null values or empty array/struct during schema inference | true, default `false` | |
+| pathFormat | choose between REGEX path format (ex: `^s3:\/\/bucket\/\/col1=B\/col2=([A-C])$`) or WILDCARD path format (ex: `s3://bucket/*/*A*/internal-*.csv`) | true, default `REGEX`| `WILDCARD` |
 
 For other options, please refer to [this doc](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/DataFrameReader.html).
 
@@ -91,6 +93,7 @@ json {
 | path | directory of parquet files | | |
 | saveMode | | true, default `ErrorIfExists` | `Append` |
 | filenamePattern | Regex of the names of file to be read. When `filenamePattern` is set, then this connector can only be used for reading data | true, default empty | `(file)(.*)(\\.csv)` |
+| pathFormat | choose between REGEX path format (ex: `^s3:\/\/bucket\/\/col1=B\/col2=([A-C])$`) or WILDCARD path format (ex: `s3://bucket/*/*A*/internal-*.csv`) | true, default `REGEX`| `WILDCARD` |
 
 ### Example
 

src/main/java/io/github/setl/enums/PathFormat.java

@@ -0,0 +1,6 @@
+package io.github.setl.enums;
+
+public enum PathFormat {
+ WILDCARD,
+ REGEX;
+}

src/main/scala/io/github/setl/config/FileConnectorConf.scala

@@ -1,7 +1,7 @@
 package io.github.setl.config
 
 import io.github.setl.annotation.InterfaceStability
-import io.github.setl.enums.Storage
+import io.github.setl.enums.{PathFormat, Storage}
 import io.github.setl.exception.ConfException
 import io.github.setl.annotation.InterfaceStability.Evolving
 import org.apache.spark.sql.SaveMode
@@ -11,6 +11,7 @@ class FileConnectorConf extends ConnectorConf {
 
  private[this] val defaultEncoding: String = "UTF-8"
  private[this] val defaultSaveMode: String = SaveMode.ErrorIfExists.toString
+ private[this] val defaultPathFormat: String = PathFormat.REGEX.toString
 
  def setStorage(storage: String): this.type = set("storage", storage.toUpperCase)
 
@@ -24,6 +25,8 @@ class FileConnectorConf extends ConnectorConf {
 
  def setPath(path: String): this.type = set("path", path)
 
+ def setPathFormat(pathFormat: PathFormat): this.type = set("pathFormat", pathFormat.toString)
+
  def setS3CredentialsProvider(value: String): this.type = set("fs.s3a.aws.credentials.provider", value)
 
  def setS3AccessKey(value: String): this.type = set("fs.s3a.access.key", value)
@@ -36,6 +39,8 @@ class FileConnectorConf extends ConnectorConf {
 
  def getSaveMode: SaveMode = SaveMode.valueOf(get("saveMode", defaultSaveMode))
 
+ def getPathFormat: String = get("pathFormat", defaultPathFormat)
+
  def getStorage: Storage = get("storage") match {
  case Some(storage) => Storage.valueOf(storage)
  case _ => throw new ConfException("The value of storage is not set")

src/main/scala/io/github/setl/storage/connector/CSVConnector.scala

@@ -2,7 +2,7 @@ package io.github.setl.storage.connector
 
 import io.github.setl.annotation.InterfaceStability
 import io.github.setl.config.{Conf, FileConnectorConf}
-import io.github.setl.enums.Storage
+import io.github.setl.enums.{PathFormat, Storage}
 import io.github.setl.util.TypesafeConfigUtils
 import com.typesafe.config.Config
 import org.apache.spark.sql._
@@ -91,12 +91,13 @@ class CSVConnector(override val options: FileConnectorConf) extends FileConnecto
 
  def this(config: Conf) = this(config.toMap)
 
- def this(path: String, inferSchema: String, delimiter: String, header: String, saveMode: SaveMode) =
+ def this(path: String, inferSchema: String, delimiter: String, header: String, saveMode: SaveMode, pathFormat: PathFormat = PathFormat.REGEX) =
  this(Map[String, String](
  "path" -> path,
  "inferSchema" -> inferSchema,
  "header" -> header,
- "saveMode" -> saveMode.toString
+ "saveMode" -> saveMode.toString,
+ "pathFormat" -> pathFormat.toString
  ))
 
  override val storage: Storage = Storage.CSV

src/main/scala/io/github/setl/storage/connector/FileConnector.scala

@@ -4,9 +4,9 @@ import java.net.{URI, URLDecoder, URLEncoder}
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.{AtomicInteger, AtomicLong}
 import java.util.concurrent.locks.ReentrantLock
-
 import io.github.setl.annotation.InterfaceStability
 import io.github.setl.config.FileConnectorConf
+import io.github.setl.enums.PathFormat
 import io.github.setl.internal.{CanDrop, CanPartition, HasReaderWriter}
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileSystem, LocalFileSystem, Path}
@@ -483,10 +483,19 @@ abstract class FileConnector(val options: FileConnectorConf) extends Connector w
  logDebug(s"Reading ${options.getStorage.toString} file in: '${absolutePath.toString}'")
  this.setJobDescription(s"Read file(s) from '${absolutePath.toString}'")
 
- val df = reader
- .option("basePath", basePath.toString)
- .format(options.getStorage.toString.toLowerCase())
- .load(listFilesToLoad(false): _*)
+ val df = options.getPathFormat match {
+ case value if value == PathFormat.WILDCARD.toString =>
+ reader
+ .format(options.getStorage.toString.toLowerCase())
+ .load(options.getPath)
+ case value if value == PathFormat.REGEX.toString =>
+ reader
+ .option("basePath", basePath.toString)
+ .format(options.getStorage.toString.toLowerCase())
+ .load(listFilesToLoad(false): _*)
+ case _ =>
+ throw new UnsupportedOperationException(s"Unsupported path format ${options.getPathFormat}.")
+ }
 
  if (dropUserDefinedSuffix & df.columns.contains(UDSKey)) {
  df.drop(UDSKey)

src/test/scala/io/github/setl/config/FileConnectorConfSuite.scala

@@ -1,6 +1,6 @@
 package io.github.setl.config
 
-import io.github.setl.enums.Storage
+import io.github.setl.enums.{PathFormat, Storage}
 import io.github.setl.exception.ConfException
 import org.apache.spark.sql.SaveMode
 import org.scalatest.funsuite.AnyFunSuite
@@ -31,6 +31,10 @@ class FileConnectorConfSuite extends AnyFunSuite {
  conf.setPath("path")
  assert(conf.get("path").get === "path")
 
+ assert(conf.get("pathFormat") === None)
+ conf.setPathFormat(PathFormat.WILDCARD)
+ assert(conf.get("pathFormat").get === "WILDCARD")
+
  assert(conf.get("credentialsProvider") === None)
  conf.setS3CredentialsProvider("credentialsProvider")
  assert(conf.get("fs.s3a.aws.credentials.provider").get === "credentialsProvider")
@@ -53,6 +57,7 @@ class FileConnectorConfSuite extends AnyFunSuite {
  assert(conf.getSaveMode === SaveMode.Overwrite)
  assert(conf.getStorage === Storage.EXCEL)
  assert(conf.getPath === "path")
+ assert(conf.getPathFormat === "WILDCARD")
  assert(conf.getSchema === None)
  assert(conf.getS3CredentialsProvider === Some("credentialsProvider"))
  assert(conf.getS3AccessKey === Some("accessKey"))

src/test/scala/io/github/setl/storage/connector/FileConnectorSuite.scala

@@ -1,7 +1,7 @@
 package io.github.setl.storage.connector
 
 import io.github.setl.config.FileConnectorConf
-import io.github.setl.enums.Storage
+import io.github.setl.enums.{PathFormat, Storage}
 import io.github.setl.{SparkSessionBuilder, TestObject}
 import org.apache.hadoop.fs.{LocalFileSystem, Path}
 import org.apache.spark.sql.{DataFrame, Dataset, SaveMode, SparkSession}
@@ -129,7 +129,7 @@ class FileConnectorSuite extends AnyFunSuite with Matchers {
  connectorRead.collect() should contain theSameElementsAs sparkRead.collect()
 
  // remove test files
- new ParquetConnector(path, SaveMode.Overwrite).delete()
+ new ParquetConnector(path, SaveMode.Overwrite).drop()
  }
 
  test("File connector should handle wildcard file path (csv)") {
@@ -163,7 +163,58 @@ class FileConnectorSuite extends AnyFunSuite with Matchers {
  connectorRead.collect() should contain theSameElementsAs sparkRead.collect()
 
  // remove test files
- new CSVConnector(path, "true", ",", "true", SaveMode.Overwrite).delete()
+ new CSVConnector(path, "true", ",", "true", SaveMode.Overwrite).drop()
+ }
+
+
+ test("File connector should handle spark native path format") {
+ val spark: SparkSession = new SparkSessionBuilder().setEnv("local").build().get()
+ val path = "src/test/resources/fileconnector_test_dir_csv"
+ import spark.implicits._
+ val df = Seq(
+ ("a", "A", "aa", "AA"),
+ ("a", "A", "bb", "BB"),
+ ("a", "B", "aa", "AA"),
+ ("a", "B", "bb", "BB"),
+ ("b", "A", "aa", "AA"),
+ ("b", "A", "bb", "BB"),
+ ("b", "B", "aa", "AA"),
+ ("b", "B", "bb", "BB")
+ ).toDF("col1", "col2", "col3", "col4")
+
+ df.write.mode(SaveMode.Overwrite).partitionBy("col1", "col2").option("header", "true").csv(path)
+
+ val connector = new CSVConnector(s"$path/col1=*/col2=A/*.csv", "true", ",", "true", SaveMode.Overwrite, PathFormat.WILDCARD)
+ val connectorRead = connector.read()
+ connectorRead.show()
+
+ val sparkRead = spark.read
+ .option("header", "true")
+ .csv(s"$path/col1=*/col2=A/*.csv")
+ sparkRead.show()
+
+ connectorRead.collect() should contain theSameElementsAs sparkRead.collect()
+
+ // remove test files
+ new CSVConnector(path, "true", ",", "true", SaveMode.Overwrite).drop()
+ }
+
+ test("FileConnector should throw exception with unsupported pathFormat option") {
+ val spark: SparkSession = new SparkSessionBuilder().setEnv("local").build().get()
+
+ import spark.implicits._
+ val options = Map[String, String](
+ "path" -> "src/test/resources",
+ "pathFormat" -> "TEST"
+ )
+ val fileConnectorConf = new FileConnectorConf()
+ fileConnectorConf.set(options)
+
+ val fileConnector = new FileConnector(fileConnectorConf) {
+ override val storage: Storage = Storage.CSV
+ }
+
+ assertThrows[UnsupportedOperationException](fileConnector.read())
  }
 
  test("File connector functionality") {
@@ -202,7 +253,7 @@ class FileConnectorSuite extends AnyFunSuite with Matchers {
  connector.write(dff.toDF, None)
  assertThrows[IllegalArgumentException](connector.write(dff.toDF, Some("test")))
  assertThrows[IllegalArgumentException](connector.setSuffix(Some("test")))
- connector.delete()
+ connector.drop()
  }
 
  test("FileConnector should handle parallel write") {
@@ -256,7 +307,7 @@ class FileConnectorSuite extends AnyFunSuite with Matchers {
  case other: Exception => throw other
  }
 
- connector.delete()
+ connector.drop()
  }
 
  test("FileConnector should handle base path correctly") {