From 05dbef7471b2d087be25909e6da5acec7bea1d39 Mon Sep 17 00:00:00 2001 From: "Sy, Andrew" Date: Fri, 18 Oct 2019 15:33:59 -0400 Subject: [PATCH 1/5] parallelizeDelimitedFileWithDdl --- .../resources/SparkFactorySuite/Delimited.by0x01.csv | 10 ++++++++++ 1 file changed, 10 insertions(+) create mode 100644 src/test/resources/SparkFactorySuite/Delimited.by0x01.csv 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 From 569079ba823d10ab7c76ad017464753765c9a1a5 Mon Sep 17 00:00:00 2001 From: "Sy, Andrew" Date: Fri, 18 Oct 2019 15:34:31 -0400 Subject: [PATCH 2/5] parallelizeDelimitedFileWithDdl --- .../msd/sparkfactory/SparkFactorySuite.scala | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) 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 From 24f5414a9d386248931228c417e047bef28d1fa9 Mon Sep 17 00:00:00 2001 From: "Sy, Andrew" Date: Fri, 18 Oct 2019 15:34:52 -0400 Subject: [PATCH 3/5] parallelizeDelimitedFileWithDdl --- pom.xml | 4 +-- .../finra/msd/sparkfactory/SparkFactory.scala | 36 +++++++++++++++++++ 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index 625eba2..237385a 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.0-SNAPSHOT jar MegaSparkDiff diff --git a/src/main/scala/org/finra/msd/sparkfactory/SparkFactory.scala b/src/main/scala/org/finra/msd/sparkfactory/SparkFactory.scala index 8a5f2f4..3dd9df5 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._ @@ -100,6 +101,41 @@ object SparkFactory { return df } + /** + * 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 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) + } + + /** * This method will create DataFrame with a single column called "field1" from a text file, the file can be on local * machine or HDFS for HDFS url like so From f7c62c21270525d78b4ac356875e19e4676ee249 Mon Sep 17 00:00:00 2001 From: "Sy, Andrew" Date: Mon, 21 Oct 2019 13:58:51 -0400 Subject: [PATCH 4/5] parallelizeDelimitedSource --- pom.xml | 2 +- .../java/org/finra/msd/enums/SourceType.java | 4 +- .../finra/msd/sparkcompare/SparkCompare.scala | 4 +- .../finra/msd/sparkfactory/SparkFactory.scala | 102 ++++++++++++------ .../DelimitedVsDelimitedSourceTest.scala | 73 +++++++++++++ 5 files changed, 147 insertions(+), 38 deletions(-) create mode 100644 src/test/scala/org/finra/msd/sparkcompare/DelimitedVsDelimitedSourceTest.scala diff --git a/pom.xml b/pom.xml index 237385a..14db468 100755 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ org.finra.megasparkdiff mega-spark-diff-asy - 0.3.0-SNAPSHOT + 0.3.1-SNAPSHOT jar MegaSparkDiff 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 3dd9df5..d5cd9de 100644 --- a/src/main/scala/org/finra/msd/sparkfactory/SparkFactory.scala +++ b/src/main/scala/org/finra/msd/sparkfactory/SparkFactory.scala @@ -101,6 +101,48 @@ object SparkFactory { return df } + /** + * This method will create DataFrame with a single column called "field1" from a text file, the file can be on local + * machine or HDFS for HDFS url like so + * "hdfs://nn1home:8020/input/war-and-peace.txt" for S3 the url like so + * "s3n://myBucket/myFile1.log" + * + * @param textFileLocation path of a flat file containing the data to be compared + * @param tempViewName temporary table name for source data + * @return custom table containing the data to be compared + */ + def parallelizeTextSource(textFileLocation: String, tempViewName: String): AppleTable = { + val df = parallelizeTextFile(textFileLocation) + val a: AppleTable = new AppleTable(SourceType.FILE, df, null, tempViewName) + return a + } + + /** This method will create an AppleTable for data in a JSON file. + * + * @param jsonFileLocation path of a json file containing the data to be compared + * @param tempViewName temporary table name for source data + * @param delimiter source data separation character + * @return custom table containing the data to be compared + */ + def parallelizeJSONSource(jsonFileLocation: String, tempViewName: String, delimiter: Option[String]): AppleTable = { + val df = sparkSession.sqlContext.read + .option("multiLine", "true") + .json(jsonFileLocation) + df.createOrReplaceTempView(tempViewName) + + AppleTable(SourceType.JSON, df, delimiter.orNull, tempViewName) + } + + /** This method will create an AppleTable for data in a JSON file. + * + * @param jsonFileLocation path of a json file containing the data to be compared + * @param tempViewName temporary table name for source data + * @return custom table containing the data to be compared + */ + def parallelizeJSONSource(jsonFileLocation: String, tempViewName: String): AppleTable = { + 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. @@ -117,6 +159,23 @@ object SparkFactory { .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. @@ -135,48 +194,23 @@ object SparkFactory { .csv(fileLocation) } - /** - * This method will create DataFrame with a single column called "field1" from a text file, the file can be on local - * machine or HDFS for HDFS url like so + * 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 textFileLocation path of a flat file containing the data to be compared - * @param tempViewName temporary table name for source data - * @return custom table containing the data to be compared - */ - def parallelizeTextSource(textFileLocation: String, tempViewName: String): AppleTable = { - val df = parallelizeTextFile(textFileLocation) - val a: AppleTable = new AppleTable(SourceType.FILE, df, null, tempViewName) - return a - } - - /** This method will create an AppleTable for data in a JSON file. - * - * @param jsonFileLocation path of a json file containing the data to be compared - * @param tempViewName temporary table name for source data - * @param delimiter source data separation character - * @return custom table containing the data to be compared + * @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 parallelizeJSONSource(jsonFileLocation: String, tempViewName: String, delimiter: Option[String]): AppleTable = { - val df = sparkSession.sqlContext.read - .option("multiLine", "true") - .json(jsonFileLocation) + def parallelizeDelimitedSourceWithDdl(fileLocation: String, ddl: String, tempViewName: String, delimiter: String = ","): AppleTable = { + val df = parallelizeDelimitedFileWithDdl(fileLocation, ddl, delimiter) df.createOrReplaceTempView(tempViewName) - - AppleTable(SourceType.JSON, df, delimiter.orNull, tempViewName) + new AppleTable(SourceType.CSV, df, delimiter, tempViewName) } - /** This method will create an AppleTable for data in a JSON file. - * - * @param jsonFileLocation path of a json file containing the data to be compared - * @param tempViewName temporary table name for source data - * @return custom table containing the data to be compared - */ - def parallelizeJSONSource(jsonFileLocation: String, tempViewName: String): AppleTable = { - parallelizeJSONSource(jsonFileLocation, tempViewName, Option.apply(",")) - } /** * This method will create an AppleTable from a query that retrieves data from a database 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 From a56cb19470ca5acb7e005a833c390da25ee3b46d Mon Sep 17 00:00:00 2001 From: "Sy, Andrew" Date: Thu, 24 Oct 2019 01:47:30 -0400 Subject: [PATCH 5/5] pom update --- pom.xml | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 14db468..4535281 100755 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ org.finra.megasparkdiff mega-spark-diff-asy - 0.3.1-SNAPSHOT + 0.3.2-SNAPSHOT jar MegaSparkDiff @@ -256,6 +256,11 @@ + + + -g:notailcalls + +