asy2 -> andrew-sy:develop

pom.xml

@@ -5,8 +5,8 @@
     <modelVersion>4.0.0</modelVersion>
 
     <groupId>org.finra.megasparkdiff</groupId>
-    <artifactId>mega-spark-diff</artifactId>
-    <version>0.3.0</version>
+    <artifactId>mega-spark-diff-asy</artifactId>
+    <version>0.3.2-SNAPSHOT</version>
     <packaging>jar</packaging>
 
     <name>MegaSparkDiff</name>
@@ -256,6 +256,11 @@
                         </goals>
                     </execution>
                 </executions>
+                <configuration>
+                    <args>
+                        <arg>-g:notailcalls</arg>
+                    </args>
+                </configuration>
             </plugin>
 
             <plugin>

src/main/java/org/finra/msd/enums/SourceType.java

@@ -24,7 +24,9 @@ public enum SourceType {
     HIVE("HIVE"),
     FILE("FILE"),
     DYNAMODB("DYNAMODB"),
-    JSON("JSON");
+    JSON("JSON"),
+    CSV("CSV"),
+    ;
 
     /**
      * Represents the source type

src/main/scala/org/finra/msd/sparkcompare/SparkCompare.scala

@@ -151,14 +151,14 @@ object SparkCompare {
     var flatRight: DataFrame = right.dataFrame
 
     //if one of the inputs has no schema then will flatten it using the delimiter
-    if (left.sourceType == SourceType.HIVE || left.sourceType == SourceType.JDBC)
+    if (left.sourceType == SourceType.HIVE || left.sourceType == SourceType.JDBC || left.sourceType == SourceType.CSV)
       flatLeft = SparkFactory.flattenDataFrame(left.dataFrame, left.delimiter)
 
     if (left.sourceType == SourceType.DYNAMODB || left.sourceType == SourceType.JSON)
       flatLeft = SparkFactory.removeComplexNullFromFlatDataFrame(
         SparkFactory.flattenDataFrame(SparkFactory.complexJSONFormatTableToSimpleJSONFormatTable(flatLeft), left.delimiter))
 
-    if (right.sourceType == SourceType.HIVE || right.sourceType == SourceType.JDBC)
+    if (right.sourceType == SourceType.HIVE || right.sourceType == SourceType.JDBC || right.sourceType == SourceType.CSV)
       flatRight = SparkFactory.flattenDataFrame(right.dataFrame, right.delimiter)
 
     if (right.sourceType == SourceType.DYNAMODB || right.sourceType == SourceType.JSON)

src/main/scala/org/finra/msd/sparkfactory/SparkFactory.scala

@@ -21,6 +21,7 @@ import org.apache.hadoop.dynamodb.DynamoDBItemWritable
 import org.apache.hadoop.dynamodb.read.DynamoDBInputFormat
 import org.apache.hadoop.io.Text
 import org.apache.hadoop.mapred.JobConf
+import org.apache.spark
 import org.apache.spark.SparkConf
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql._
@@ -142,6 +143,75 @@ object SparkFactory {
     parallelizeJSONSource(jsonFileLocation, tempViewName, Option.apply(","))
   }
 
+  /**
+    * Create DataFrame from a delimited text file.
+    * The file can be on local machine or HDFS or other file system supported by the Spark implementation.
+    * "hdfs://nn1home:8020/input/war-and-peace.txt" for S3 the url like so
+    * "s3n://myBucket/myFile1.log"
+    *
+    * @param fileLocation path of file
+    * @param delimiter delimiter used in the delimited file
+    * @return DataFrame created from the file
+    */
+  def parallelizeDelimitedFile(fileLocation: String, delimiter: String = ","): DataFrame = {
+    sparkSession.read
+      .option("delimiter", delimiter)
+      .csv(fileLocation)
+  }
+
+  /**
+    * Create AppleTable from a delimited text file.
+    * The file can be on local machine or HDFS or other file system supported by the Spark implementation.
+    * "hdfs://nn1home:8020/input/war-and-peace.txt" for S3 the url like so
+    * "s3n://myBucket/myFile1.log"
+    *
+    * @param fileLocation path of file
+    * @param delimiter delimiter used in the delimited file
+    * @param tempViewName    temporary table name for source data
+    * @return AppleTable created from the file
+    */
+  def parallelizeDelimitedSource(fileLocation: String, tempViewName: String, delimiter: String = ","): AppleTable = {
+    val df = parallelizeDelimitedFile(fileLocation, delimiter)
+    df.createOrReplaceTempView(tempViewName)
+    new AppleTable(SourceType.CSV, df, delimiter, tempViewName)
+  }
+
+  /**
+    * Create DataFrame from a delimited text file, applying the specified schema.
+    * The file can be on local machine or HDFS or other file system supported by the Spark implementation.
+    * "hdfs://nn1home:8020/input/war-and-peace.txt" for S3 the url like so
+    * "s3n://myBucket/myFile1.log"
+    *
+    * @param fileLocation path of file
+    * @param delimiter delimiter used in the delimited file
+    * @param ddl schema specified in DDL style, e.g. "name VARCHAR(50), age INT"
+    * @return DataFrame created from the file
+    */
+  def parallelizeDelimitedFileWithDdl(fileLocation: String, ddl: String, delimiter: String = ","): DataFrame = {
+    sparkSession.read
+      .option("delimiter", delimiter)
+      .schema(ddl)
+      .csv(fileLocation)
+  }
+
+  /**
+    * Create AppleTable from a delimited text file.
+    * The file can be on local machine or HDFS or other file system supported by the Spark implementation.
+    * "hdfs://nn1home:8020/input/war-and-peace.txt" for S3 the url like so
+    * "s3n://myBucket/myFile1.log"
+    *
+    * @param fileLocation path of file
+    * @param delimiter delimiter used in the delimited file
+    * @param tempViewName    temporary table name for source data
+    * @return AppleTable created from the file
+    */
+  def parallelizeDelimitedSourceWithDdl(fileLocation: String, ddl: String, tempViewName: String, delimiter: String = ","): AppleTable = {
+    val df = parallelizeDelimitedFileWithDdl(fileLocation, ddl, delimiter)
+    df.createOrReplaceTempView(tempViewName)
+    new AppleTable(SourceType.CSV, df, delimiter, tempViewName)
+  }
+
+
   /**
     * This method will create an AppleTable from a query that retrieves data from a database
     * accessed through JDBC connection.

src/test/resources/SparkFactorySuite/Delimited.by0x01.csv

@@ -0,0 +1,10 @@
+43000023
ER973231
Publish
2019-10-14 06:21:11.897185
CUSIP9
+43000052
ER973231
Publish
2019-10-14 06:21:31.007094
CUSIP9
+43000077
ER973232
Publish
2019-10-14 06:38:02.640739
CUSIP9
+43000107
ER973232
Publish
2019-10-14 06:38:04.406405
CUSIP9
+43000163
ER973234
Publish
2019-10-14 09:05:34.569691
CUSIP9
+43000224
ER973234
Publish
2019-10-14 09:05:38.132313
CUSIP9
+43000280
ER973235
Publish
2019-10-14 09:09:38.409236
CUSIP9
+43000341
ER973235
Publish
2019-10-14 09:09:42.237475
CUSIP9
+43000396
ER973236
Publish
2019-10-14 09:11:15.452184
CUSIP9
+43000457
ER973236
Publish
2019-10-14 09:11:17.311592
CUSIP9

src/test/scala/org/finra/msd/sparkcompare/DelimitedVsDelimitedSourceTest.scala

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2017 MegaSparkDiff Contributors
+ *
+ * Licensed 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.
+ *//*
+ * Copyright 2017 MegaSparkDiff Contributors
+ *
+ * Licensed 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.finra.msd.sparkcompare
+
+import org.finra.msd.basetestclasses.SparkFunSuite
+import org.finra.msd.sparkfactory.SparkFactory
+
+class DelimitedVsDelimitedSourceTest() extends SparkFunSuite {
+  private def returnDiff(fileName1: String, fileName2: String) = {
+    val file1Path = this.getClass.getClassLoader.getResource(fileName1).getPath
+    val leftAppleTable = SparkFactory.parallelizeDelimitedSource(file1Path, "table1")
+    val file2Path = this.getClass.getClassLoader.getResource(fileName2).getPath
+    val rightAppleTable = SparkFactory.parallelizeDelimitedSource(file2Path, "table2")
+    SparkCompare.compareAppleTables(leftAppleTable, rightAppleTable)
+  }
+
+  test("testCompareEqualFiles") {
+    val diffResult = returnDiff("Test1.txt", "Test2.txt")
+    assert(diffResult.inLeftNotInRight.count == 0)
+    assert(diffResult.inRightNotInLeft.count == 0)
+  }
+
+  test("testCompareCompletelyDifferentFiles") {
+    val diffResult = returnDiff("Test4.txt", "Test5.txt")
+    assert(diffResult.inLeftNotInRight.count == 5)
+    assert(diffResult.inRightNotInLeft.count == 4)
+  }
+
+  test("testCompareAFewDifferences") {
+    val diffResult = returnDiff("Test1.txt", "Test3.txt")
+    assert(diffResult.inLeftNotInRight.count == 2)
+    assert(diffResult.inRightNotInLeft.count == 2)
+  }
+
+  test("testCompareTable1IsSubset") {
+    val diffResult = returnDiff("Test4.txt", "Test1.txt")
+    assert(diffResult.inLeftNotInRight.count == 0)
+    assert(diffResult.inRightNotInLeft.count == 4)
+  }
+
+  test("testCompareTable2IsSubset") {
+    val diffResult = returnDiff("Test1.txt", "Test5.txt")
+    assert(diffResult.inLeftNotInRight.count == 5)
+    assert(diffResult.inRightNotInLeft.count == 0)
+  }
+}

src/test/scala/org/finra/msd/sparkfactory/SparkFactorySuite.scala

@@ -44,4 +44,36 @@ class SparkFactorySuite extends SparkFunSuite {
     val rightAppleTable = SparkFactory.parallelizeJDBCSource("org.hsqldb.jdbc.JDBCDriver", "jdbc:hsqldb:hsql://127.0.0.1:9001/testDb", "SA", "", "(select * from Test1 )", "my_partition_test", scala.Option.empty, "Price", "0", "200000", "2")
     if (rightAppleTable.getDataFrame.rdd.getNumPartitions != 2) fail("expected 2 partitions but received " + rightAppleTable.getDataFrame.rdd.getNumPartitions)
   }
+
+
+  test("parallelizeDelimitedFile can load file delimited by default delimiter")
+  {
+    val filePath = this.getClass.getClassLoader.getResource( "Test1.txt").getPath
+    val dataFrame = SparkFactory.parallelizeDelimitedFile(filePath)
+    assert (dataFrame.columns.size == 4)
+  }
+
+  test("parallelizeDelimitedFile can load file delimited by u0001")
+  {
+    val filePath = this.getClass.getClassLoader.getResource( "SparkFactorySuite/Delimited.by0x01.csv").getPath
+    val dataFrame = SparkFactory.parallelizeDelimitedFile(filePath, "\u0001")
+    assert (dataFrame.columns.size == 5)
+  }
+
+  test("parallelizeDelimitedFileWithDdl can apply specified schema")
+  {
+    val schema =
+      """
+        |    `SBMSN_ID` INT,
+        |    `SBMSN_IDNTR` VARCHAR(25),
+        |    `SBMSN_ST` VARCHAR(25),
+        |    `SBMSN_DT` TIMESTAMP,
+        |    `SCRTY_TYPE` VARCHAR(100)
+        |""".stripMargin
+
+    val filePath = this.getClass.getClassLoader.getResource( "SparkFactorySuite/Delimited.by0x01.csv").getPath
+    val df = SparkFactory.parallelizeDelimitedFileWithDdl(filePath, schema, "\u0001")
+    assert (df.columns.sameElements(Array("SBMSN_ID", "SBMSN_IDNTR", "SBMSN_ST", "SBMSN_DT", "SCRTY_TYPE")))
+  }
+
 }