From e36972cc987c2008507ae61265b9296f87e71dbe Mon Sep 17 00:00:00 2001 From: changyonglik Date: Tue, 12 Jul 2022 00:44:50 +0900 Subject: [PATCH 1/9] Added Feathr config doc --- docs/how-to-guides/feathr-job-configuration.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/how-to-guides/feathr-job-configuration.md b/docs/how-to-guides/feathr-job-configuration.md index 70a2e9505..5967642a9 100644 --- a/docs/how-to-guides/feathr-job-configuration.md +++ b/docs/how-to-guides/feathr-job-configuration.md @@ -12,3 +12,4 @@ Since Feathr uses Spark as the underlying execution engine, there's a way to ove | ------------------------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------- | | spark.feathr.inputFormat | None | Specify the input format if the file cannot be tell automatically. By default, Feathr will read files by parsing the file extension name; However the file/folder name doesn't have extension name, this configuration can be set to tell Feathr which format it should use to read the data. Currently can only be set for Spark built-in short names, including `json`, `parquet`, `jdbc`, `orc`, `libsvm`, `csv`, `text`. For more details, see ["Manually Specifying Options"](https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html#manually-specifying-options). Additionally, `delta` is also supported if users want to read delta lake. | 0.2.1 | | spark.feathr.outputFormat | None | Specify the output format. "avro" is the default behavior if this value is not set. Currently can only be set for Spark built-in short names, including `json`, `parquet`, `jdbc`, `orc`, `libsvm`, `csv`, `text`. For more details, see ["Manually Specifying Options"](https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html#manually-specifying-options). Additionally, `delta` is also supported if users want to write delta lake. | 0.2.1 | +| spark.feathr.inputFormat.csvOptions.sep | None | Specify the delimiter. For example, "," for commas or "\t" for tabs. | TODO: Check version | \ No newline at end of file From 27d9a2b03713d1a579fa4f941a5d216bb2c060aa Mon Sep 17 00:00:00 2001 From: changyonglik Date: Tue, 12 Jul 2022 20:16:27 +0900 Subject: [PATCH 2/9] Added csvDelimiterOption to batchdataloader Signed-off-by: changyonglik --- .../feathr/offline/source/dataloader/BatchDataLoader.scala | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/BatchDataLoader.scala b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/BatchDataLoader.scala index db430fb4a..386d44627 100644 --- a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/BatchDataLoader.scala +++ b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/BatchDataLoader.scala @@ -64,6 +64,10 @@ private[offline] class BatchDataLoader(ss: SparkSession, location: InputLocation val dataPath = location.getPath log.info(s"Loading ${location} as DataFrame, using parameters ${dataIOParametersWithSplitSize}") + + val sqlContext = ss.sqlContext + val csvDelimiterOption = sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep", ",") + try { import scala.util.control.Breaks._ @@ -88,7 +92,7 @@ private[offline] class BatchDataLoader(ss: SparkSession, location: InputLocation throw feathrException // Throwing exception to avoid dataLoaderHandler hook exception from being swallowed. case e: Throwable => //TODO: Analyze all thrown exceptions, instead of swalling them all, and reading as a csv println(e.toString) - ss.read.format("csv").option("header", "true").load(dataPath) + ss.read.format("csv").option("header", "true").option("delimiter", csvDelimiterOption).load(dataPath) } } } From bacdadbe51d6652702b0742fd92094f8c38c9a32 Mon Sep 17 00:00:00 2001 From: changyonglik Date: Tue, 12 Jul 2022 20:55:44 +0900 Subject: [PATCH 3/9] Added unit test for batchdataloader Signed-off-by: changyonglik --- src/test/resources/anchor1-source.tsv | 8 +++ .../dataloader/TestBatchDataLoader.scala | 52 +++++++++++++++++++ 2 files changed, 60 insertions(+) create mode 100644 src/test/resources/anchor1-source.tsv create mode 100644 src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala diff --git a/src/test/resources/anchor1-source.tsv b/src/test/resources/anchor1-source.tsv new file mode 100644 index 000000000..176ed963d --- /dev/null +++ b/src/test/resources/anchor1-source.tsv @@ -0,0 +1,8 @@ +mId alpha beta gamma omega +1 apple 10 10 0.1 +2 orange 10 3 0.1 +3 banana 10 2 0.9 +4 apple 10 1 0.7 +5 apple 11 11 1.0 +7 banana 2 10 81.27 +9 banana 4 4 0.4 \ No newline at end of file diff --git a/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala new file mode 100644 index 000000000..9efc5b436 --- /dev/null +++ b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala @@ -0,0 +1,52 @@ +package com.linkedin.feathr.offline.source.dataloader + +import com.linkedin.feathr.offline.TestFeathr +import com.linkedin.feathr.offline.config.location.SimplePath +import org.apache.spark.sql.Row +import org.testng.Assert.assertEquals +import org.testng.annotations.Test + +/** + * unit tests for [[BatchDataLoader]] + */ +class TestBatchDataLoader extends TestFeathr { + + @Test(description = "test loading dataframe with BatchDataLoader") + def testBatchDataLoader() : Unit = { + val path = "anchor1-source.csv" + val absolutePath = getClass.getClassLoader.getResource(path).getPath + val batchDataLoader = new BatchDataLoader(ss, SimplePath(absolutePath), List()) + val df = batchDataLoader.loadDataFrame() + val expectedRows = Array( + Row("1", "apple", "10", "10", "0.1"), + Row("2", "orange", "10", "3", "0.1"), + Row("3", "banana", "10", "2", "0.9"), + Row("4", "apple", "10", "1", "0.7"), + Row("5", "apple", "11", "11", "1.0"), + Row("7", "banana", "2", "10", "81.27"), + Row("9", "banana", "4", "4", "0.4") + ) + assertEquals(df.collect(), expectedRows) + } + + @Test(description = "test loading dataframe with BatchDataLoader by specifying delimiter") + def testBatchDataLoaderWithCsvDelimiterOption() : Unit = { + val path = "anchor1-source.tsv" + val absolutePath = getClass.getClassLoader.getResource(path).getPath + val sqlContext = ss.sqlContext + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "\t") + val batchDataLoader = new BatchDataLoader(ss, SimplePath(absolutePath), List()) + val df = batchDataLoader.loadDataFrame() + val expectedRows = Array( + Row("1", "apple", "10", "10", "0.1"), + Row("2", "orange", "10", "3", "0.1"), + Row("3", "banana", "10", "2", "0.9"), + Row("4", "apple", "10", "1", "0.7"), + Row("5", "apple", "11", "11", "1.0"), + Row("7", "banana", "2", "10", "81.27"), + Row("9", "banana", "4", "4", "0.4") + ) + assertEquals(df.collect(), expectedRows) + } + +} From d55e384a44ffe7b86d079fb44e875bcae6ed05e9 Mon Sep 17 00:00:00 2001 From: changyonglik Date: Tue, 12 Jul 2022 23:32:18 +0900 Subject: [PATCH 4/9] Added csvDelimiterOption to Fileformat Signed-off-by: changyonglik --- .../offline/source/dataloader/hdfs/FileFormat.scala | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/FileFormat.scala b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/FileFormat.scala index 925e649cc..8ad9f4829 100644 --- a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/FileFormat.scala +++ b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/FileFormat.scala @@ -51,9 +51,11 @@ object FileFormat { // TODO: Complete a general loadDataFrame and replace current adhoc load data frame code def loadDataFrame(ss: SparkSession, path: String, format: String = CSV): DataFrame = { + val sqlContext = ss.sqlContext + val csvDelimiterOption = sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep", ",") format match { case AVRO => new AvroJsonDataLoader(ss, path).loadDataFrame() - case CSV => ss.read.format("csv").option("header", "true").load(path) + case CSV => ss.read.format("csv").option("header", "true").option("delimiter", csvDelimiterOption).load(path) case PARQUET => new ParquetDataLoader(ss, path).loadDataFrame() case _ => ??? } @@ -69,23 +71,26 @@ object FileFormat { val p = existingHdfsPaths.head.toLowerCase() p match { case p if p.endsWith(".csv") => CSV + case p if p.endsWith(".tsv") => CSV case p if p.endsWith(".parquet") => PARQUET case p if p.endsWith(".orc") => ORC case p if p.endsWith(".avro.json") => AVRO_JSON case p if p.endsWith(".avro") => AVRO case p if p.startsWith("jdbc:") => JDBC case _ => - // if we cannot tell the file format from the file extensions, we should read from `spark.feathr.inputFormat` to get the format that's sepcified by user. - dataIOParameters.getOrElse(DATA_FORMAT, ss.conf.get("spark.feathr.inputFormat", AVRO)).toUpperCase + // if we cannot tell the file format from the file extensions, we should read from `spark.feathr.inputFormat` to get the format that's sepcified by user. + dataIOParameters.getOrElse(DATA_FORMAT, ss.conf.get("spark.feathr.inputFormat", AVRO)).toUpperCase } } def loadHdfsDataFrame(format: String, existingHdfsPaths: Seq[String]): DataFrame = { + val sqlContext = ss.sqlContext + val csvDelimiterOption = sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep", ",") val df = format match { case CSV => - ss.read.format("csv").option("header", "true").load(existingHdfsPaths: _*) + ss.read.format("csv").option("header", "true").option("delimiter", csvDelimiterOption).load(existingHdfsPaths: _*) case AVRO => ss.read.format(AVRO_DATASOURCE).load(existingHdfsPaths: _*) case ORC => From 804a85c9e2176bcea5e19e260f53cc55c240852f Mon Sep 17 00:00:00 2001 From: changyonglik Date: Tue, 12 Jul 2022 23:55:52 +0900 Subject: [PATCH 5/9] Added unittest for Fileformat Signed-off-by: changyonglik --- .../source/dataloader/BatchDataLoader.scala | 15 +++- .../source/dataloader/hdfs/FileFormat.scala | 25 ++++++- .../dataloader/TestBatchDataLoader.scala | 7 ++ .../dataloader/hdfs/TestFileFormat.scala | 73 +++++++++++++++++++ 4 files changed, 117 insertions(+), 3 deletions(-) create mode 100644 src/test/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/TestFileFormat.scala diff --git a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/BatchDataLoader.scala b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/BatchDataLoader.scala index 386d44627..5251480b5 100644 --- a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/BatchDataLoader.scala +++ b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/BatchDataLoader.scala @@ -51,6 +51,15 @@ private[offline] class BatchDataLoader(ss: SparkSession, location: InputLocation loadDataFrame(Map(), new JobConf(ss.sparkContext.hadoopConfiguration)) } + /** + * Convert string to special characters + * @return a String + */ + def escape(raw: String): String = { + import scala.reflect.runtime.universe._ + Literal(Constant(raw)).toString.replaceAll("\"", "") + } + /** * load the source data as dataframe. * @param dataIOParameters extra parameters @@ -65,8 +74,12 @@ private[offline] class BatchDataLoader(ss: SparkSession, location: InputLocation log.info(s"Loading ${location} as DataFrame, using parameters ${dataIOParametersWithSplitSize}") + // Get csvDelimiterOption set with spark.feathr.inputFormat.csvOptions.sep val sqlContext = ss.sqlContext - val csvDelimiterOption = sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep", ",") + // Get rawCsvDelimiterOption from spark.feathr.inputFormat.csvOptions.sep + val rawCsvDelimiterOption = sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep", ",") + // If rawCsvDelimiterOption is not properly set, defaults to "," as the delimiter else csvDelimiterOption + val csvDelimiterOption = if (escape(rawCsvDelimiterOption).trim.isEmpty) "," else rawCsvDelimiterOption try { import scala.util.control.Breaks._ diff --git a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/FileFormat.scala b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/FileFormat.scala index 8ad9f4829..9e48812e7 100644 --- a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/FileFormat.scala +++ b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/FileFormat.scala @@ -29,6 +29,15 @@ object FileFormat { val DATA_FORMAT = "data.format" + /** + * Convert string to special characters + * @return a String + */ + def escape(raw: String): String = { + import scala.reflect.runtime.universe._ + Literal(Constant(raw)).toString.replaceAll("\"", "") + } + /** * To define if the file is JDBC, Single File or Path list (default) * @param path @@ -51,8 +60,14 @@ object FileFormat { // TODO: Complete a general loadDataFrame and replace current adhoc load data frame code def loadDataFrame(ss: SparkSession, path: String, format: String = CSV): DataFrame = { + + // Get csvDelimiterOption set with spark.feathr.inputFormat.csvOptions.sep val sqlContext = ss.sqlContext - val csvDelimiterOption = sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep", ",") + // Get rawCsvDelimiterOption from spark.feathr.inputFormat.csvOptions.sep + val rawCsvDelimiterOption = sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep", ",") + // If rawCsvDelimiterOption is not properly set, defaults to "," as the delimiter else csvDelimiterOption + val csvDelimiterOption = if (escape(rawCsvDelimiterOption).trim.isEmpty) "," else rawCsvDelimiterOption + format match { case AVRO => new AvroJsonDataLoader(ss, path).loadDataFrame() case CSV => ss.read.format("csv").option("header", "true").option("delimiter", csvDelimiterOption).load(path) @@ -86,8 +101,14 @@ object FileFormat { } def loadHdfsDataFrame(format: String, existingHdfsPaths: Seq[String]): DataFrame = { + + // Get csvDelimiterOption set with spark.feathr.inputFormat.csvOptions.sep val sqlContext = ss.sqlContext - val csvDelimiterOption = sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep", ",") + // Get rawCsvDelimiterOption from spark.feathr.inputFormat.csvOptions.sep + val rawCsvDelimiterOption = sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep", ",") + // If rawCsvDelimiterOption is not properly set, defaults to "," as the delimiter else csvDelimiterOption + val csvDelimiterOption = if (escape(rawCsvDelimiterOption).trim.isEmpty) "," else rawCsvDelimiterOption + val df = format match { case CSV => ss.read.format("csv").option("header", "true").option("delimiter", csvDelimiterOption).load(existingHdfsPaths: _*) diff --git a/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala index 9efc5b436..77301be0b 100644 --- a/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala +++ b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala @@ -11,6 +11,11 @@ import org.testng.annotations.Test */ class TestBatchDataLoader extends TestFeathr { + def escape(raw: String): String = { + import scala.reflect.runtime.universe._ + Literal(Constant(raw)).toString + } + @Test(description = "test loading dataframe with BatchDataLoader") def testBatchDataLoader() : Unit = { val path = "anchor1-source.csv" @@ -35,6 +40,7 @@ class TestBatchDataLoader extends TestFeathr { val absolutePath = getClass.getClassLoader.getResource(path).getPath val sqlContext = ss.sqlContext sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "\t") + println(s"Postset config testBatchDataLoaderWithCsvDelimiterOption: ${ss.sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep")}") val batchDataLoader = new BatchDataLoader(ss, SimplePath(absolutePath), List()) val df = batchDataLoader.loadDataFrame() val expectedRows = Array( @@ -47,6 +53,7 @@ class TestBatchDataLoader extends TestFeathr { Row("9", "banana", "4", "4", "0.4") ) assertEquals(df.collect(), expectedRows) + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep","") } } diff --git a/src/test/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/TestFileFormat.scala b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/TestFileFormat.scala new file mode 100644 index 000000000..7016bffc5 --- /dev/null +++ b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/hdfs/TestFileFormat.scala @@ -0,0 +1,73 @@ +package com.linkedin.feathr.offline.source.dataloader.hdfs + +import com.linkedin.feathr.offline.TestFeathr +import org.apache.spark.sql.{Row, SparkSession} +import org.testng.Assert.assertEquals +import org.testng.annotations.Test + +/** + * unit tests for [[FileFormat]] + */ +class TestFileFormat extends TestFeathr { + + @Test(description = "test loading dataframe with FileFormat") + def testLoadDataFrame() : Unit = { + val path = "anchor1-source.csv" + val absolutePath = getClass.getClassLoader.getResource(path).getPath + val df = FileFormat.loadDataFrame(ss, absolutePath, "CSV") + + val expectedRows = Array( + Row("1", "apple", "10", "10", "0.1"), + Row("2", "orange", "10", "3", "0.1"), + Row("3", "banana", "10", "2", "0.9"), + Row("4", "apple", "10", "1", "0.7"), + Row("5", "apple", "11", "11", "1.0"), + Row("7", "banana", "2", "10", "81.27"), + Row("9", "banana", "4", "4", "0.4") + ) + assertEquals(df.collect(), expectedRows) + } + + @Test(description = "test loading dataframe with FileFormat by specifying delimiter") + def testLoadDataFrameWithCsvDelimiterOption() : Unit = { + val path = "anchor1-source.tsv" + val absolutePath = getClass.getClassLoader.getResource(path).getPath + val sqlContext = ss.sqlContext + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "\t") + val df = FileFormat.loadDataFrame(ss, absolutePath, "CSV") + + val expectedRows = Array( + Row("1", "apple", "10", "10", "0.1"), + Row("2", "orange", "10", "3", "0.1"), + Row("3", "banana", "10", "2", "0.9"), + Row("4", "apple", "10", "1", "0.7"), + Row("5", "apple", "11", "11", "1.0"), + Row("7", "banana", "2", "10", "81.27"), + Row("9", "banana", "4", "4", "0.4") + ) + assertEquals(df.collect(), expectedRows) + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "") + } + + @Test(description = "test loading dataframe with FileFormat by specifying delimiter for HDFS") + def testLoadDataFrameWithCsvDelimiterOptionHDFS() : Unit = { + val path = "anchor1-source.tsv" + val absolutePath = getClass.getClassLoader.getResource(path).getPath + val sqlContext = ss.sqlContext + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "\t") + val df = FileFormat.loadHdfsDataFrame("CSV", Seq(absolutePath)) + + val expectedRows = Array( + Row("1", "apple", "10", "10", "0.1"), + Row("2", "orange", "10", "3", "0.1"), + Row("3", "banana", "10", "2", "0.9"), + Row("4", "apple", "10", "1", "0.7"), + Row("5", "apple", "11", "11", "1.0"), + Row("7", "banana", "2", "10", "81.27"), + Row("9", "banana", "4", "4", "0.4") + ) + assertEquals(df.collect(), expectedRows) + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "") + } + +} \ No newline at end of file From cac59a0422db4d9254c89855661d34700bbde678 Mon Sep 17 00:00:00 2001 From: changyonglik Date: Fri, 22 Jul 2022 23:58:51 +0900 Subject: [PATCH 6/9] Added unittest for Fileformat Signed-off-by: changyonglik --- .../feathr/offline/source/dataloader/TestBatchDataLoader.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala index 77301be0b..02cf54dd0 100644 --- a/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala +++ b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestBatchDataLoader.scala @@ -53,7 +53,7 @@ class TestBatchDataLoader extends TestFeathr { Row("9", "banana", "4", "4", "0.4") ) assertEquals(df.collect(), expectedRows) - sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep","") + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "") } } From bb6901ea37595c66c2503345f072be2a600c92b2 Mon Sep 17 00:00:00 2001 From: changyonglik Date: Sat, 23 Jul 2022 00:20:37 +0900 Subject: [PATCH 7/9] Added csvDataLoader and unit test Signed-off-by: changyonglik --- .../source/dataloader/CsvDataLoader.scala | 22 +++++++++++++++++-- .../source/dataloader/TestCsvDataLoader.scala | 21 +++++++++++++++++- 2 files changed, 40 insertions(+), 3 deletions(-) diff --git a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/CsvDataLoader.scala b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/CsvDataLoader.scala index 4e5b53012..510c680cd 100644 --- a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/CsvDataLoader.scala +++ b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/CsvDataLoader.scala @@ -27,22 +27,40 @@ private[offline] class CsvDataLoader(ss: SparkSession, path: String) extends Dat doLoadCsvDocumentLikeAvro()._2 } + /** + * Convert string to special characters + * @return a String + */ + def escape(raw: String): String = { + import scala.reflect.runtime.universe._ + Literal(Constant(raw)).toString.replaceAll("\"", "") + } + + /** * load the source data as dataframe. * @return an dataframe */ override def loadDataFrame(): DataFrame = { + + // Get csvDelimiterOption set with spark.feathr.inputFormat.csvOptions.sep + val sqlContext = ss.sqlContext + // Get rawCsvDelimiterOption from spark.feathr.inputFormat.csvOptions.sep + val rawCsvDelimiterOption = sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep", ",") + // If rawCsvDelimiterOption is not properly set, defaults to "," as the delimiter else csvDelimiterOption + val csvDelimiterOption = if (escape(rawCsvDelimiterOption).trim.isEmpty) "," else rawCsvDelimiterOption + try { log.debug(s"Loading CSV path :${path}") val absolutePath = new File(path).getPath log.debug(s"Got absolute CSV path: ${absolutePath}, loading..") - ss.read.format("csv").option("header", "true").load(absolutePath) + ss.read.format("csv").option("header", "true").option("delimiter", csvDelimiterOption).load(absolutePath) } catch { case _: Throwable => log.debug(s"Loading CSV failed, retry with class loader..") val absolutePath = getClass.getClassLoader.getResource(path).getPath log.debug(s"Got absolution CSV path from class loader: ${absolutePath}, loading.. ") - ss.read.format("csv").option("header", "true").load(absolutePath) + ss.read.format("csv").option("header", "true").option("delimiter", csvDelimiterOption).load(absolutePath) } } diff --git a/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestCsvDataLoader.scala b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestCsvDataLoader.scala index f89782661..cee7ab935 100644 --- a/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestCsvDataLoader.scala +++ b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestCsvDataLoader.scala @@ -29,7 +29,6 @@ class TestCsvDataLoader extends TestFeathr { assertEquals(df.collect(), expectedRows) } - @Test(description = "test loading Avro schema with CsvDataLoader") def testLoadSchema() : Unit = { val dataLoader = new CsvDataLoader(ss, "anchor1-source.csv") @@ -46,4 +45,24 @@ class TestCsvDataLoader extends TestFeathr { val expectedSchema = Schema.createRecord(expectedFields) assertEquals(schema.getFields, expectedSchema.getFields) } + + @Test(description = "test loading dataframe with CsvDataLoader by specifying delimiter") + def testLoadDataFrameWithCsvDelimiterOption() : Unit = { + val sqlContext = ss.sqlContext + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "\t") + val dataLoader = new CsvDataLoader(ss, "anchor1-source.tsv") + + val df = dataLoader.loadDataFrame() + val expectedRows = Array( + Row("1", "apple", "10", "10", "0.1"), + Row("2", "orange", "10", "3", "0.1"), + Row("3", "banana", "10", "2", "0.9"), + Row("4", "apple", "10", "1", "0.7"), + Row("5", "apple", "11", "11", "1.0"), + Row("7", "banana", "2", "10", "81.27"), + Row("9", "banana", "4", "4", "0.4") + ) + assertEquals(df.collect(), expectedRows) + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "") + } } From 45fb65f163595498b235a1776bca562423bd3917 Mon Sep 17 00:00:00 2001 From: changyonglik Date: Sat, 23 Jul 2022 00:46:33 +0900 Subject: [PATCH 8/9] Added sourceUtils and unit test Signed-off-by: changyonglik --- .../feathr/offline/util/SourceUtils.scala | 19 +++++- .../feathr/offline/util/TestSourceUtils.scala | 59 +++++++++++++++++++ 2 files changed, 77 insertions(+), 1 deletion(-) create mode 100644 src/test/scala/com/linkedin/feathr/offline/util/TestSourceUtils.scala diff --git a/src/main/scala/com/linkedin/feathr/offline/util/SourceUtils.scala b/src/main/scala/com/linkedin/feathr/offline/util/SourceUtils.scala index 95838d1a0..bcea562fb 100644 --- a/src/main/scala/com/linkedin/feathr/offline/util/SourceUtils.scala +++ b/src/main/scala/com/linkedin/feathr/offline/util/SourceUtils.scala @@ -65,6 +65,15 @@ private[offline] object SourceUtils { val FEATURE_MP_DEF_CONFIG_SUFFIX = ".conf" val firstRecordName = "topLevelRecord" + /** + * Convert string to special characters + * @return a String + */ + def escape(raw: String): String = { + import scala.reflect.runtime.universe._ + Literal(Constant(raw)).toString.replaceAll("\"", "") + } + /** * get AVRO datum type of a dataset we should use to load, * it is determined by the expect datatype from a set of anchor transformers @@ -665,6 +674,14 @@ private[offline] object SourceUtils { // TODO: Split isLocal case into Test Packages val format = FileFormat.getType(inputData.inputPath) log.info(s"loading ${inputData.inputPath} input Path as Format: ${format}") + + // Get csvDelimiterOption set with spark.feathr.inputFormat.csvOptions.sep + val sqlContext = ss.sqlContext + // Get rawCsvDelimiterOption from spark.feathr.inputFormat.csvOptions.sep + val rawCsvDelimiterOption = sqlContext.getConf("spark.feathr.inputFormat.csvOptions.sep", ",") + // If rawCsvDelimiterOption is not properly set, defaults to "," as the delimiter else csvDelimiterOption + val csvDelimiterOption = if (escape(rawCsvDelimiterOption).trim.isEmpty) "," else rawCsvDelimiterOption + format match { case FileFormat.PATHLIST => { val pathList = getPathList(sourceFormatType=inputData.sourceType, @@ -689,7 +706,7 @@ private[offline] object SourceUtils { JdbcUtils.loadDataFrame(ss, inputData.inputPath) } case FileFormat.CSV => { - ss.read.format("csv").option("header", "true").load(inputData.inputPath) + ss.read.format("csv").option("header", "true").option("delimiter", csvDelimiterOption).load(inputData.inputPath) } case _ => { if (ss.sparkContext.isLocal){ diff --git a/src/test/scala/com/linkedin/feathr/offline/util/TestSourceUtils.scala b/src/test/scala/com/linkedin/feathr/offline/util/TestSourceUtils.scala new file mode 100644 index 000000000..0ce2df9a6 --- /dev/null +++ b/src/test/scala/com/linkedin/feathr/offline/util/TestSourceUtils.scala @@ -0,0 +1,59 @@ +package com.linkedin.feathr.offline.util + +import com.linkedin.feathr.offline.TestFeathr +import com.linkedin.feathr.offline.client.InputData +import com.linkedin.feathr.offline.source.SourceFormatType +import org.apache.hadoop.conf.Configuration +import org.apache.spark.sql.Row +import org.testng.Assert.assertEquals +import org.testng.annotations.Test + +/** + * unit tests for [[SourceUtils]] + */ +class TestSourceUtils extends TestFeathr { + + @Test(description = "test loading dataframe with SourceUtils") + def testSourceUtils() : Unit = { + val conf: Configuration = new Configuration() + val path = "anchor1-source.csv" + val absolutePath = getClass.getClassLoader.getResource(path).getPath + val inputData = InputData(absolutePath, SourceFormatType.FIXED_PATH) + val df = SourceUtils.loadObservationAsDF(ss, conf, inputData, List()) + + val expectedRows = Array( + Row("1", "apple", "10", "10", "0.1"), + Row("2", "orange", "10", "3", "0.1"), + Row("3", "banana", "10", "2", "0.9"), + Row("4", "apple", "10", "1", "0.7"), + Row("5", "apple", "11", "11", "1.0"), + Row("7", "banana", "2", "10", "81.27"), + Row("9", "banana", "4", "4", "0.4") + ) + assertEquals(df.collect(), expectedRows) + } + + @Test(description = "test loading dataframe with SourceUtils by specifying delimiter") + def testSourceUtilsWithCsvDelimiterOption() : Unit = { + val conf: Configuration = new Configuration() + val path = "anchor1-source.tsv" + val sqlContext = ss.sqlContext + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "\t") + val absolutePath = getClass.getClassLoader.getResource(path).getPath + val inputData = InputData(absolutePath, SourceFormatType.FIXED_PATH) + val df = SourceUtils.loadObservationAsDF(ss, conf, inputData, List()) + + val expectedRows = Array( + Row("1", "apple", "10", "10", "0.1"), + Row("2", "orange", "10", "3", "0.1"), + Row("3", "banana", "10", "2", "0.9"), + Row("4", "apple", "10", "1", "0.7"), + Row("5", "apple", "11", "11", "1.0"), + Row("7", "banana", "2", "10", "81.27"), + Row("9", "banana", "4", "4", "0.4") + ) + assertEquals(df.collect(), expectedRows) + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "") + } + +} \ No newline at end of file From 5b52fd6ea341dc63d2df0a328fcad6217717e13b Mon Sep 17 00:00:00 2001 From: changyonglik Date: Sat, 23 Jul 2022 01:07:10 +0900 Subject: [PATCH 9/9] Added test for BatchLoader in TestDataLoaderFactory Signed-off-by: changyonglik --- .../offline/source/dataloader/TestDataLoaderFactory.scala | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestDataLoaderFactory.scala b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestDataLoaderFactory.scala index 1ee8c8d19..fd62992c6 100644 --- a/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestDataLoaderFactory.scala +++ b/src/test/scala/com/linkedin/feathr/offline/source/dataloader/TestDataLoaderFactory.scala @@ -14,6 +14,11 @@ class TestDataLoaderFactory extends TestFeathr { val localDataLoaderFactory = new LocalDataLoaderFactory(ss, dataLoaderHandlers=List()) val csvLoader = localDataLoaderFactory.create("anchor1-source.csv") assertTrue(csvLoader.isInstanceOf[CsvDataLoader]) + val sqlContext = ss.sqlContext + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "\t") + val csvLoaderWithDelimiter = localDataLoaderFactory.create("anchor1-source.tsv") + assertTrue(csvLoaderWithDelimiter.isInstanceOf[BatchDataLoader]) + sqlContext.setConf("spark.feathr.inputFormat.csvOptions.sep", "") val avroJsonLoader = localDataLoaderFactory.create("anchor5-source.avro.json") assertTrue(avroJsonLoader.isInstanceOf[AvroJsonDataLoader]) val jsonWithSchemaLoader = localDataLoaderFactory.create("simple-obs2") // the mock data folder exists.