Repository: G-Research/spark-extension (language: python)

Issue:
Add Diff app
Add an Spark application that diffs two files from hadoop filesystem and outputs statistics on stdout and diff to file.

This allows to use the package as a stand-alone Spark application ready to run on a cluster, e.g. as a step in a regression test pipeline.


Discussion / hints:

Fix patch:
diff --git a/.github/actions/build/action.yml b/.github/actions/build/action.yml
index e73b096e..8c1194d7 100644
--- a/.github/actions/build/action.yml
+++ b/.github/actions/build/action.yml
@@ -50,6 +50,7 @@ runs:
         *
         !.*
         !target/*-javadoc.jar
+        !target/*-sources.jar
         !target/site
 
 branding:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index ec7a9dea..624b9074 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -182,39 +182,49 @@ jobs:
             scala-compat-version: '2.12'
             scala-version: '2.12.10'
             spark-patch-version: '3'
+            hadoop-version: '2.7'
           - spark-compat-version: '3.1'
             scala-compat-version: '2.12'
             scala-version: '2.12.10'
             spark-patch-version: '3'
+            hadoop-version: '2.7'
           - spark-compat-version: '3.2'
             scala-compat-version: '2.12'
             scala-version: '2.12.15'
+            hadoop-version: '2.7'
           - spark-compat-version: '3.2'
             scala-compat-version: '2.12'
             scala-version: '2.12.15'
             spark-patch-version: '4'
+            hadoop-version: '2.7'
           - spark-compat-version: '3.3'
             scala-compat-version: '2.12'
             scala-version: '2.12.15'
+            hadoop-version: '2'
           - spark-compat-version: '3.4'
             scala-compat-version: '2.12'
             scala-version: '2.12.17'
             spark-patch-version: '0'
+            hadoop-version: '3'
 
           - spark-compat-version: '3.2'
             scala-compat-version: '2.13'
             scala-version: '2.13.5'
+            hadoop-version: '3.2'
           - spark-compat-version: '3.2'
             scala-compat-version: '2.13'
             scala-version: '2.13.5'
             spark-patch-version: '4'
+            hadoop-version: '3.2'
           - spark-compat-version: '3.3'
             scala-compat-version: '2.13'
             scala-version: '2.13.8'
+            hadoop-version: '3'
           - spark-compat-version: '3.4'
             scala-compat-version: '2.13'
             scala-version: '2.13.8'
             spark-patch-version: '0'
+            hadoop-version: '3'
 
     steps:
     - name: Checkout
@@ -229,6 +239,7 @@ jobs:
         scala-version: ${{ matrix.scala-version }}
         spark-compat-version: ${{ matrix.spark-compat-version }}
         scala-compat-version: ${{ matrix.scala-compat-version }}
+        hadoop-version: ${{ matrix.hadoop-version }}
 
   # pyspark is not available for snapshots or scala other than 2.12
   # we would have to compile spark from sources for this, not worth it
diff --git a/CHANGELOG.md b/CHANGELOG.md
index dc422d22..7799a1f3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,11 @@
 All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
+## [UNRELEASED] - YYYY-MM-DD
+
+### Added
+
+- Spark app to diff files or tables and write result back to file or table. (#160)
 
 ## [2.6.0] - 2023-04-11
 
diff --git a/DIFF.md b/DIFF.md
index df8ebc9b..1c03a0f6 100644
--- a/DIFF.md
+++ b/DIFF.md
@@ -403,3 +403,85 @@ The latter variant is prefixed with `_with_options`.
 
 * `def diff(self: DataFrame, other: DataFrame, *id_columns: str) -> DataFrame`
 * `def diffwith(self: DataFrame, other: DataFrame, *id_columns: str) -> DataFrame:`
+
+## Diff Spark application
+
+There is also a Spark application that can be used to create a diff DataFrame. The application reads two DataFrames
+`left` and `right` from files or tables, executes the diff transformation and writes the result DataFrame to a file or table.
+The Diff app can be run via `spark-submit`:
+
+```shell
+# Scala 2.12
+spark-submit --packages com.github.scopt:scopt_2.12:4.1.0 spark-extension_2.12-2.7.0-3.4.jar --help
+
+# Scala 2.13
+spark-submit --packages com.github.scopt:scopt_2.13:4.1.0 spark-extension_2.13-2.7.0-3.4.jar --help
+```
+
+```
+Spark Diff app (2.7.0-3.4)
+
+Usage: spark-extension_2.13-2.7.0-3.4.jar [options] left right diff
+
+  left                     file path (requires format option) or table name to read left dataframe
+  right                    file path (requires format option) or table name to read right dataframe
+  diff                     file path (requires format option) or table name to write diff dataframe
+
+Examples:
+
+  - Diff CSV files 'left.csv' and 'right.csv' and write result into CSV file 'diff.csv':
+    spark-submit --packages com.github.scopt:scopt_2.13:4.1.0 spark-extension_2.13-2.7.0-3.4.jar --format csv left.csv right.csv diff.csv
+
+  - Diff CSV file 'left.csv' with Parquet file 'right.parquet' with id column 'id', and write result into Hive table 'diff':
+    spark-submit --packages com.github.scopt:scopt_2.13:4.1.0 spark-extension_2.13-2.7.0-3.4.jar --left-format csv --right-format parquet --hive --id id left.csv right.parquet diff
+
+Spark session
+  --master <master>        Spark master (local, yarn, ...), not needed with spark-submit
+  --app-name <app-name>    Spark application name
+  --hive                   enable Hive support to read from and write to Hive tables
+
+Input and output
+  -f, --format <format>    input and output file format (csv, json, parquet, ...)
+  --left-format <format>   left input file format (csv, json, parquet, ...)
+  --right-format <format>  right input file format (csv, json, parquet, ...)
+  --output-format <formt>  output file format (csv, json, parquet, ...)
+
+  -s, --schema <schema>    input schema
+  --left-schema <schema>   left input schema
+  --right-schema <schema>  right input schema
+
+  --left-option:key=val    left input option
+  --right-option:key=val   right input option
+  --output-option:key=val  output option
+
+  --id <name>              id column name
+  --ignore <name>          ignore column name
+  --save-mode <save-mode>  save mode for writing output (Append, Overwrite, ErrorIfExists, Ignore, default ErrorIfExists)
+
+Diffing options
+  --diff-column <name>     column name for diff column (default 'diff')
+  --left-prefix <prefix>   prefix for left column names (default 'left')
+  --right-prefix <prefix>  prefix for right column names (default 'right')
+  --insert-value <value>   value for insertion (default 'I')
+  --change-value <value>   value for change (default 'C')
+  --delete-value <value>   value for deletion (default 'D')
+  --no-change-value <val>  value for no change (default 'N')
+  --change-column <name>   column name for change column (default is no such column)
+  --diff-mode <mode>       diff mode (ColumnByColumn, SideBySide, LeftSide, RightSide, default ColumnByColumn)
+  --sparse                 enable sparse diff
+
+General
+  --help                   prints this usage text
+```
+
+### Examples
+
+Diff CSV files `left.csv` and `right.csv` and write result into CSV file `diff.csv`:
+```shell
+spark-submit --packages com.github.scopt:scopt_2.13:4.1.0 spark-extension_2.13-2.7.0-3.4.jar --format csv left.csv right.csv diff.csv
+```
+
+Diff CSV file `left.csv` with Parquet file `right.parquet` with id column `id`, and write result into Hive table `diff`:
+```shell
+spark-submit --packages com.github.scopt:scopt_2.13:4.1.0 spark-extension_2.13-2.7.0-3.4.jar --left-format csv --right-format parquet --hive --id id left.csv right.parquet diff
+```
diff --git a/README.md b/README.md
index cd3a6e4d..2ae1c750 100644
--- a/README.md
+++ b/README.md
@@ -2,7 +2,7 @@
 
 This project provides extensions to the [Apache Spark project](https://spark.apache.org/) in Scala and Python:
 
-**[Diff](DIFF.md):** A `diff` transformation for `Dataset`s that computes the differences between
+**[Diff](DIFF.md):** A `diff` transformation and application for `Dataset`s that computes the differences between
 two datasets, i.e. which rows to _add_, _delete_ or _change_ to get from one dataset to the other.
 
 **[SortedGroups](GROUPS.md):** A `groupByKey` transformation that groups rows by a key while providing
diff --git a/pom.xml b/pom.xml
index 5999bc92..724040c9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -72,6 +72,20 @@
       <scope>provided</scope>
     </dependency>
 
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-hive_${scala.compat.version}</artifactId>
+      <version>${spark.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>com.github.scopt</groupId>
+      <artifactId>scopt_${scala.compat.version}</artifactId>
+      <!-- keep DIFF.md section "Diff Spark application" synced with this value -->
+      <version>4.1.0</version>
+    </dependency>
+
     <!-- Test -->
     <dependency>
       <groupId>org.apache.spark</groupId>
@@ -209,6 +223,19 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-jar-plugin</artifactId>
+        <version>3.3.0</version>
+        <configuration>
+          <archive>
+            <manifest>
+              <addClasspath>true</addClasspath>
+              <mainClass>uk.co.gresearch.spark.diff.App</mainClass>
+            </manifest>
+          </archive>
+        </configuration>
+      </plugin>
       <!-- run java tests -->
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
diff --git a/src/main/scala/uk/co/gresearch/spark/diff/App.scala b/src/main/scala/uk/co/gresearch/spark/diff/App.scala
new file mode 100644
index 00000000..09e5b8a7
--- /dev/null
+++ b/src/main/scala/uk/co/gresearch/spark/diff/App.scala
@@ -0,0 +1,250 @@
+package uk.co.gresearch.spark.diff
+
+import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
+import scopt.OptionParser
+import uk.co.gresearch._
+
+object App {
+  // define available options
+  case class Options(master: Option[String] = None,
+                     appName: Option[String] = None,
+                     hive: Boolean = false,
+
+                     leftPath: Option[String] = None,
+                     rightPath: Option[String] = None,
+                     outputPath: Option[String] = None,
+
+                     leftFormat: Option[String] = None,
+                     rightFormat: Option[String] = None,
+                     outputFormat: Option[String] = None,
+
+                     leftSchema: Option[String] = None,
+                     rightSchema: Option[String] = None,
+
+                     leftOptions: Map[String, String] = Map.empty,
+                     rightOptions: Map[String, String] = Map.empty,
+                     outputOptions: Map[String, String] = Map.empty,
+
+                     ids: Seq[String] = Seq.empty,
+                     ignore: Seq[String] = Seq.empty,
+                     saveMode: SaveMode = SaveMode.ErrorIfExists,
+                     diffOptions: DiffOptions = DiffOptions.default)
+
+  // read options from args
+  val programName = s"spark-extension_${spark.BuildScalaCompatVersionString}-${spark.VersionString}.jar"
+  val parser: OptionParser[Options] = new scopt.OptionParser[Options](programName) {
+    head(s"Spark Diff app (${spark.VersionString})")
+    head()
+
+    arg[String]("left")
+      .required()
+      .valueName("<left-path>")
+      .action((x, c) => c.copy(leftPath = Some(x)))
+      .text("file path (requires format option) or table name to read left dataframe")
+
+    arg[String]("right")
+      .required()
+      .valueName("<right-path>")
+      .action((x, c) => c.copy(rightPath = Some(x)))
+      .text("file path (requires format option) or table name to read right dataframe")
+
+    arg[String]("diff")
+      .required()
+      .valueName("<diff-path>")
+      .action((x, c) => c.copy(outputPath = Some(x)))
+      .text("file path (requires format option) or table name to write diff dataframe")
+
+    note("")
+    note("Examples:")
+    note("")
+    note("  - Diff CSV files 'left.csv' and 'right.csv' and write result into CSV file 'diff.csv':")
+    note(s"    spark-submit --packages com.github.scopt:scopt_${spark.BuildScalaCompatVersionString}:4.1.0 $programName --format csv left.csv right.csv diff.csv")
+    note("")
+    note("  - Diff CSV file 'left.csv' with Parquet file 'right.parquet' with id column 'id', and write result into Hive table 'diff':")
+    note(s"    spark-submit --packages com.github.scopt:scopt_${spark.BuildScalaCompatVersionString}:4.1.0 $programName --left-format csv --right-format parquet --hive --id id left.csv right.parquet diff")
+
+    note("")
+    note("Spark session")
+    opt[String]("master")
+      .valueName("<master>")
+      .action((x, c) => c.copy(master = Some(x)))
+      .text("Spark master (local, yarn, ...), not needed with spark-submit")
+    opt[String]("app-name")
+      .valueName("<app-name>")
+      .action((x, c) => c.copy(appName = Some(x)))
+      .text("Spark application name")
+      .withFallback(() => "Diff App")
+    opt[Unit]("hive")
+      .optional()
+      .action((_, c) => c.copy(hive = true))
+      .text(s"enable Hive support to read from and write to Hive tables")
+
+    note("")
+    note("Input and output")
+    opt[String]('f', "format")
+      .valueName("<format>")
+      .action((x, c) => c.copy(
+        leftFormat = c.leftFormat.orElse(Some(x)),
+        rightFormat = c.rightFormat.orElse(Some(x)),
+        outputFormat = c.outputFormat.orElse(Some(x))
+      ))
+      .text("input and output file format (csv, json, parquet, ...)")
+    opt[String]("left-format")
+      .valueName("<format>")
+      .action((x, c) => c.copy(leftFormat = Some(x)))
+      .text("left input file format (csv, json, parquet, ...)")
+    opt[String]("right-format")
+      .valueName("<format>")
+      .action((x, c) => c.copy(rightFormat = Some(x)))
+      .text("right input file format (csv, json, parquet, ...)")
+    opt[String]("output-format")
+      .valueName("<formt>")
+      .action((x, c) => c.copy(outputFormat = Some(x)))
+      .text("output file format (csv, json, parquet, ...)")
+
+    note("")
+    opt[String]('s', "schema")
+      .valueName("<schema>")
+      .action((x, c) => c.copy(
+        leftSchema = c.leftSchema.orElse(Some(x)),
+        rightSchema = c.rightSchema.orElse(Some(x))
+      ))
+      .text("input schema")
+    opt[String]("left-schema")
+      .valueName("<schema>")
+      .action((x, c) => c.copy(leftSchema = Some(x)))
+      .text("left input schema")
+    opt[String]("right-schema")
+      .valueName("<schema>")
+      .action((x, c) => c.copy(rightSchema = Some(x)))
+      .text("right input schema")
+
+    note("")
+    opt[(String, String)]("left-option")
+      .unbounded()
+      .optional()
+      .keyValueName("key", "val")
+      .action((x, c) => c.copy(leftOptions = c.leftOptions + (x._1 -> x._2)))
+      .text("left input option")
+    opt[(String, String)]("right-option")
+      .unbounded()
+      .optional()
+      .keyValueName("key", "val")
+      .action((x, c) => c.copy(rightOptions = c.rightOptions + (x._1 -> x._2)))
+      .text("right input option")
+    opt[(String, String)]("output-option")
+      .unbounded()
+      .optional()
+      .keyValueName("key", "val")
+      .action((x, c) => c.copy(outputOptions = c.outputOptions + (x._1 -> x._2)))
+      .text("output option")
+
+    note("")
+    opt[String]("id")
+      .unbounded()
+      .valueName("<name>")
+      .action((x, c) => c.copy(ids = c.ids :+ x))
+      .text(s"id column name")
+    opt[String]("ignore")
+      .unbounded()
+      .valueName("<name>")
+      .action((x, c) => c.copy(ignore = c.ignore :+ x))
+      .text(s"ignore column name")
+    opt[String]("save-mode")
+      .optional()
+      .valueName("<save-mode>")
+      .action((x, c) => c.copy(saveMode = SaveMode.valueOf(x)))
+      .text(s"save mode for writing output (${SaveMode.values().mkString(", ")}, default ${Options().saveMode})")
+
+    note("")
+    note("Diffing options")
+    opt[String]("diff-column")
+      .optional()
+      .valueName("<name>")
+      .action((x, c) => c.copy(diffOptions = c.diffOptions.copy(diffColumn = x)))
+      .text(s"column name for diff column (default '${DiffOptions.default.diffColumn}')")
+    opt[String]("left-prefix")
+      .optional()
+      .valueName("<prefix>")
+      .action((x, c) => c.copy(diffOptions = c.diffOptions.copy(leftColumnPrefix = x)))
+      .text(s"prefix for left column names (default '${DiffOptions.default.leftColumnPrefix}')")
+    opt[String]("right-prefix")
+      .optional()
+      .valueName("<prefix>")
+      .action((x, c) => c.copy(diffOptions = c.diffOptions.copy(rightColumnPrefix = x)))
+      .text(s"prefix for right column names (default '${DiffOptions.default.rightColumnPrefix}')")
+    opt[String]("insert-value")
+      .optional()
+      .valueName("<value>")
+      .action((x, c) => c.copy(diffOptions = c.diffOptions.copy(insertDiffValue = x)))
+      .text(s"value for insertion (default '${DiffOptions.default.insertDiffValue}')")
+    opt[String]("change-value")
+      .optional()
+      .valueName("<value>")
+      .action((x, c) => c.copy(diffOptions = c.diffOptions.copy(changeDiffValue = x)))
+      .text(s"value for change (default '${DiffOptions.default.changeDiffValue}')")
+    opt[String]("delete-value")
+      .optional()
+      .valueName("<value>")
+      .action((x, c) => c.copy(diffOptions = c.diffOptions.copy(deleteDiffValue = x)))
+      .text(s"value for deletion (default '${DiffOptions.default.deleteDiffValue}')")
+    opt[String]("no-change-value")
+      .optional()
+      .valueName("<val>")
+      .action((x, c) => c.copy(diffOptions = c.diffOptions.copy(nochangeDiffValue = x)))
+      .text(s"value for no change (default '${DiffOptions.default.nochangeDiffValue}')")
+    opt[String]("change-column")
+      .optional()
+      .valueName("<name>")
+      .action((x, c) => c.copy(diffOptions = c.diffOptions.copy(changeColumn = Some(x))))
+      .text(s"column name for change column (default is no such column)")
+    opt[String]("diff-mode")
+      .optional()
+      .valueName("<mode>")
+      .action((x, c) => c.copy(diffOptions = c.diffOptions.copy(diffMode = DiffMode.withName(x))))
+      .text(s"diff mode (${DiffMode.values.mkString(", ")}, default ${Options().diffOptions.diffMode})")
+    opt[Unit]("sparse")
+      .optional()
+      .action((_, c) => c.copy(diffOptions = c.diffOptions.copy(sparseMode = true)))
+      .text(s"enable sparse diff")
+
+    note("")
+    note("General")
+    help("help").text("prints this usage text")
+  }
+
+  def read(spark: SparkSession, format: Option[String], path: String, schema: Option[String], options: Map[String, String]): DataFrame =
+    spark.read
+      .when(format.isDefined).call(_.format(format.get))
+      .options(options)
+      .when(schema.isDefined).call(_.schema(schema.get))
+      .when(format.isDefined).either(_.load(path)).or(_.table(path))
+
+  def write(df: DataFrame, format: Option[String], path: String, options: Map[String, String], saveMode: SaveMode): Unit =
+    df.write
+      .when(format.isDefined).call(_.format(format.get))
+      .options(options)
+      .mode(saveMode)
+      .when(format.isDefined).either(_.save(path)).or(_.saveAsTable(path))
+
+  def main(args: Array[String]): Unit = {
+    // parse options
+    val options = parser.parse(args, Options()) match {
+      case Some(options) => options
+      case None => sys.exit(1)
+    }
+
+    // create spark session
+    val spark = SparkSession.builder()
+      .appName(options.appName.get)
+      .when(options.hive).call(_.enableHiveSupport())
+      .when(options.master.isDefined).call(_.master(options.master.get))
+      .getOrCreate()
+
+    // read and write
+    val left = read(spark, options.leftFormat, options.leftPath.get, options.leftSchema, options.leftOptions)
+    val right = read(spark, options.rightFormat, options.rightPath.get, options.rightSchema, options.rightOptions)
+    val diff = left.diff(right, options.diffOptions, options.ids, options.ignore)
+    write(diff, options.outputFormat, options.outputPath.get, options.outputOptions, options.saveMode)
+  }
+}


Test patch:
diff --git a/.github/actions/test-jvm/action.yml b/.github/actions/test-jvm/action.yml
index 40108acf..4ab8d551 100644
--- a/.github/actions/test-jvm/action.yml
+++ b/.github/actions/test-jvm/action.yml
@@ -31,6 +31,26 @@ runs:
       name: Binaries-${{ inputs.spark-compat-version }}-${{ inputs.scala-version }}
       path: .
 
+  - name: Cache Spark Binaries
+    uses: actions/cache@v3
+    if: ( ! contains(inputs.spark-version, '-SNAPSHOT') )
+    with:
+      path: ~/spark
+      key: ${{ runner.os }}-spark-binaries-${{ inputs.spark-version }}-${{ inputs.scala-compat-version }}
+
+  - name: Setup Spark Binaries
+    if: ( ! contains(inputs.spark-version, '-SNAPSHOT') )
+    env:
+      SPARK_PACKAGE: spark-${{ inputs.spark-version }}/spark-${{ inputs.spark-version }}-bin-hadoop${{ inputs.hadoop-version }}${{ inputs.scala-compat-version == '2.13' && '-scala2.13' || '' }}.tgz
+    run: |
+      if [[ ! -e ~/spark ]]
+      then
+        wget --progress=dot:giga "https://www.apache.org/dyn/closer.lua/spark/${SPARK_PACKAGE}?action=download" -O - | tar -xzC "${{ runner.temp }}"
+        archive=$(basename "${SPARK_PACKAGE}") bash -c "mv -v "${{ runner.temp }}/\${archive/%.tgz/}" ~/spark"
+      fi
+      echo "SPARK_HOME=$(cd ~/spark; pwd)" >> $GITHUB_ENV
+    shell: bash
+
   - name: Cache Maven packages
     uses: actions/cache@v3
     with:
@@ -48,6 +68,13 @@ runs:
     run: mvn --batch-mode test
     shell: bash
 
+  - name: Diff App test
+    if: ( ! contains(inputs.spark-version, '-SNAPSHOT') )
+    run: |
+      $SPARK_HOME/bin/spark-submit --packages com.github.scopt:scopt_${{ inputs.scala-compat-version }}:4.1.0 target/spark-extension_*.jar --format parquet --id id src/test/files/test.parquet/file1.parquet src/test/files/test.parquet/file2.parquet diff.parquet
+      $SPARK_HOME/bin/spark-shell <<< 'val df = spark.read.parquet("diff.parquet").orderBy($"id").groupBy($"diff").count; df.show; if (df.count != 2) sys.exit(1)'
+    shell: bash
+
   - name: Generate Unit Test Report
     if: failure()
     run: mvn --batch-mode surefire-report:report-only
diff --git a/src/test/scala/uk/co/gresearch/spark/SparkTestSession.scala b/src/test/scala/uk/co/gresearch/spark/SparkTestSession.scala
index 80cccc9c..f9b79bd5 100644
--- a/src/test/scala/uk/co/gresearch/spark/SparkTestSession.scala
+++ b/src/test/scala/uk/co/gresearch/spark/SparkTestSession.scala
@@ -29,6 +29,7 @@ trait SparkTestSession extends SQLHelper {
       .appName("spark test example")
       .config("spark.sql.shuffle.partitions", 2)
       .config("spark.local.dir", ".")
+      .enableHiveSupport()
       .getOrCreate()
   }
 
diff --git a/src/test/scala/uk/co/gresearch/spark/diff/AppSuite.scala b/src/test/scala/uk/co/gresearch/spark/diff/AppSuite.scala
new file mode 100644
index 00000000..cdd2e2b9
--- /dev/null
+++ b/src/test/scala/uk/co/gresearch/spark/diff/AppSuite.scala
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2023 G-Research
+ *
+ * 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 uk.co.gresearch.spark.diff
+
+import org.apache.spark.sql.SaveMode
+import org.scalatest.funsuite.AnyFunSuite
+import uk.co.gresearch.spark.SparkTestSession
+
+import java.io.File
+
+
+class AppSuite extends AnyFunSuite with SparkTestSession {
+  import spark.implicits._
+
+  test("run app with file and hive table") {
+    withTempPath { path =>
+      // write left dataframe as csv
+      val leftPath = new File(path, "left.csv").getAbsolutePath
+      DiffSuite.left(spark).write.csv(leftPath)
+
+      // write right dataframe as parquet table
+      DiffSuite.right(spark).write.format("parquet").mode(SaveMode.Overwrite).saveAsTable("right_parquet")
+
+      // launch app
+      val jsonPath = new File(path, "diff.json").getAbsolutePath
+      App.main(Array(
+        "--left-format", "csv",
+        "--left-schema", "id int, value string",
+        "--output-format", "json",
+        "--id", "id",
+        leftPath,
+        "right_parquet",
+        jsonPath
+      ))
+
+      // assert written diff
+      val actual = spark.read.json(jsonPath)
+      assert(actual.orderBy($"id").collect() === DiffSuite.expectedDiff)
+    }
+  }
+}
diff --git a/src/test/scala/uk/co/gresearch/spark/diff/DiffSuite.scala b/src/test/scala/uk/co/gresearch/spark/diff/DiffSuite.scala
index b277b3b6..c47551c0 100644
--- a/src/test/scala/uk/co/gresearch/spark/diff/DiffSuite.scala
+++ b/src/test/scala/uk/co/gresearch/spark/diff/DiffSuite.scala
@@ -19,7 +19,7 @@ package uk.co.gresearch.spark.diff
 import org.apache.spark.sql.functions.regexp_replace
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
-import org.apache.spark.sql.{Dataset, Encoders, Row}
+import org.apache.spark.sql.{Dataset, Encoders, Row, SparkSession}
 import org.scalatest.funsuite.AnyFunSuite
 import uk.co.gresearch.spark.{SparkTestSession, distinctPrefixFor}
 
@@ -93,21 +93,40 @@ case class DiffAsOneSide(diff: String,
                          id: Int,
                          value: Option[String])
 
+object DiffSuite {
+  def left(spark: SparkSession): Dataset[Value] = {
+    import spark.implicits._
+    Seq(
+      Value(1, Some("one")),
+      Value(2, Some("two")),
+      Value(3, Some("three"))
+    ).toDS()
+  }
+
+  def right(spark: SparkSession): Dataset[Value] = {
+    import spark.implicits._
+    Seq(
+      Value(1, Some("one")),
+      Value(2, Some("Two")),
+      Value(4, Some("four"))
+    ).toDS()
+  }
+
+  val expectedDiff: Seq[Row] = Seq(
+    Row("N", 1, "one", "one"),
+    Row("C", 2, "two", "Two"),
+    Row("D", 3, "three", null),
+    Row("I", 4, null, "four")
+  )
+
+}
+
 class DiffSuite extends AnyFunSuite with SparkTestSession {
 
   import spark.implicits._
 
-  lazy val left: Dataset[Value] = Seq(
-    Value(1, Some("one")),
-    Value(2, Some("two")),
-    Value(3, Some("three"))
-  ).toDS()
-
-  lazy val right: Dataset[Value] = Seq(
-    Value(1, Some("one")),
-    Value(2, Some("Two")),
-    Value(4, Some("four"))
-  ).toDS()
+  lazy val left: Dataset[Value] = DiffSuite.left(spark)
+  lazy val right: Dataset[Value] = DiffSuite.right(spark)
 
   lazy val left7: Dataset[Value7] = Seq(
     Value7(1, Some("one"), Some("one label")),
@@ -155,12 +174,7 @@ class DiffSuite extends AnyFunSuite with SparkTestSession {
 
   lazy val expectedDiffColumns: Seq[String] = Seq("diff", "id", "left_value", "right_value")
 
-  lazy val expectedDiff: Seq[Row] = Seq(
-    Row("N", 1, "one", "one"),
-    Row("C", 2, "two", "Two"),
-    Row("D", 3, "three", null),
-    Row("I", 4, null, "four")
-  )
+  lazy val expectedDiff: Seq[Row] = DiffSuite.expectedDiff
 
   lazy val expectedReverseDiff: Seq[Row] = Seq(
     Row("N", 1, "one", "one"),