diff --git a/pom.xml b/pom.xml index 625eba2..4535281 100755 --- a/pom.xml +++ b/pom.xml @@ -5,8 +5,8 @@ 4.0.0 org.finra.megasparkdiff - mega-spark-diff - 0.3.0 + mega-spark-diff-asy + 0.3.2-SNAPSHOT jar MegaSparkDiff @@ -256,6 +256,11 @@ + + + -g:notailcalls + + diff --git a/src/main/java/org/finra/msd/enums/SourceType.java b/src/main/java/org/finra/msd/enums/SourceType.java index 5f363b7..630e56f 100644 --- a/src/main/java/org/finra/msd/enums/SourceType.java +++ b/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 diff --git a/src/main/scala/org/finra/msd/sparkcompare/SparkCompare.scala b/src/main/scala/org/finra/msd/sparkcompare/SparkCompare.scala index 27b275e..1375eed 100755 --- a/src/main/scala/org/finra/msd/sparkcompare/SparkCompare.scala +++ b/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) diff --git a/src/main/scala/org/finra/msd/sparkfactory/SparkFactory.scala b/src/main/scala/org/finra/msd/sparkfactory/SparkFactory.scala index 8a5f2f4..d5cd9de 100644 --- a/src/main/scala/org/finra/msd/sparkfactory/SparkFactory.scala +++ b/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. diff --git a/src/test/resources/SparkFactorySuite/Delimited.by0x01.csv b/src/test/resources/SparkFactorySuite/Delimited.by0x01.csv new file mode 100644 index 0000000..67ff479 --- /dev/null +++ b/src/test/resources/SparkFactorySuite/Delimited.by0x01.csv @@ -0,0 +1,10 @@ +43000023ER973231Publish2019-10-14 06:21:11.897185CUSIP9 +43000052ER973231Publish2019-10-14 06:21:31.007094CUSIP9 +43000077ER973232Publish2019-10-14 06:38:02.640739CUSIP9 +43000107ER973232Publish2019-10-14 06:38:04.406405CUSIP9 +43000163ER973234Publish2019-10-14 09:05:34.569691CUSIP9 +43000224ER973234Publish2019-10-14 09:05:38.132313CUSIP9 +43000280ER973235Publish2019-10-14 09:09:38.409236CUSIP9 +43000341ER973235Publish2019-10-14 09:09:42.237475CUSIP9 +43000396ER973236Publish2019-10-14 09:11:15.452184CUSIP9 +43000457ER973236Publish2019-10-14 09:11:17.311592CUSIP9 diff --git a/src/test/scala/org/finra/msd/sparkcompare/DelimitedVsDelimitedSourceTest.scala b/src/test/scala/org/finra/msd/sparkcompare/DelimitedVsDelimitedSourceTest.scala new file mode 100644 index 0000000..8489819 --- /dev/null +++ b/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) + } +} \ No newline at end of file diff --git a/src/test/scala/org/finra/msd/sparkfactory/SparkFactorySuite.scala b/src/test/scala/org/finra/msd/sparkfactory/SparkFactorySuite.scala index ecec68e..eddeb3b 100644 --- a/src/test/scala/org/finra/msd/sparkfactory/SparkFactorySuite.scala +++ b/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"))) + } + } \ No newline at end of file