diff --git a/README.md b/README.md index 3142f7e5..083e69ed 100644 --- a/README.md +++ b/README.md @@ -3138,6 +3138,49 @@ pramen.operations = [ ``` + +### (Experimental) Bulk load of historical data +Sometimes data needs to be loaded and processed for historical periods that may span several years. You can, of course, +run Pramen with `--date-from` and `--date-to` to load such data, but for daily datasets this can take a long time because +each day is processed independently. Bulk loading allows data to be loaded in `monthly`, `quarterly`, or `yearly` chunks. + +Bulk loading works as follows: +- Each month, quarter, or year is loaded into a single info date partition corresponding to the first info date of that + period. +- Data is loaded for each period independently. Pramen tracks progress in the `bulk_loads` table. If a job is interrupted + and later restarted, Pramen resumes processing from where it left off. +- After all data for a period has been loaded, you can enable repartitioning so that `pramen_info_date` matches the + daily dates. + +Example configuration options: +```hocon +pramen { + # The period of data to load + load.date.from = "2000-01-01" + load.date.to = "2020-12-31" + + runtime.run.mode = bulk + runtime.run.bulk.batch.size = monthly # Can be quarterly or yearly as well + runtime.inverse.order = true + + runtime.info.date.column = "transaction_timestamp" + runtime.info.date.format = "yyyy-MM-dd" # Only is the info date column data type is not 'date' or 'timestamp' + runtime.enable.repartitioning = true # This is false by default - please use with caustion since this is an experimental feature +} +``` + +Alternatively, you can use command line to run bulk loads without changing the config like this: +``` +--workflow "dummy.config" \ +--date-from "2000-01-01" \ +--date-to "2020-12-31" \ +--inverse-order "true" \ +--run-mode "bulk" \ +--bulk-size "yearly" \ +--info-date-column "info_date" \ +--info-date-format "yyyyMMdd" +``` + ## Pipeline Notifications Custom pipeline notification targets allow execution arbitrary actions after the pipeline is finished. Usually, it is used to send custom notifications to external systems. A pipeline notification target can be created by implementing diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/BulkRunConfig.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/BulkRunConfig.scala index 7e4c1a06..84ea0970 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/BulkRunConfig.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/BulkRunConfig.scala @@ -22,5 +22,6 @@ case class BulkRunConfig( dataDateFrom: LocalDate, dataDateTo: LocalDate, infoDateColumn: Option[String], + infoDateFormat: String, outputInfoDate: LocalDate ) diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/RuntimeConfig.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/RuntimeConfig.scala index 101b08e8..4a404c1c 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/RuntimeConfig.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/RuntimeConfig.scala @@ -69,6 +69,7 @@ object RuntimeConfig { val RUN_BULK_BATCH_SIZE = "pramen.runtime.bulk.batch.size" val RUN_ENABLE_REPARTITIONING = "pramen.runtime.enable.repartitioning" val INFO_DATE_COLUMN = "pramen.runtime.info.date.column" + val INFO_DATE_FORMAT = "pramen.runtime.info.date.format" val BULK_CURRENT_DATE_FROM = "pramen.runtime.run.bulk.current.date.from" val BULK_CURRENT_DATE_TO = "pramen.runtime.run.bulk.current.date.to" val BULK_CURRENT_OUTPUT_INFO_DATE = "pramen.runtime.run.bulk.current.output.information.date" @@ -165,10 +166,11 @@ object RuntimeConfig { val bulkCurrentDateTo = ConfigUtils.getOptionString(conf, BULK_CURRENT_DATE_TO).map(getDate) val bulkCurrentOutputInfoDate = ConfigUtils.getOptionString(conf, BULK_CURRENT_OUTPUT_INFO_DATE).map(getDate) val infoDateColumn = ConfigUtils.getOptionString(conf, INFO_DATE_COLUMN) + val infoDateFormat2 = ConfigUtils.getOptionString(conf, INFO_DATE_FORMAT).getOrElse(infoDateFormat) val enableRepartitioning = ConfigUtils.getOptionBoolean(conf, RUN_ENABLE_REPARTITIONING).getOrElse(false) val bulkLoadCurrent = if (bulkCurrentDateFrom.isDefined && bulkCurrentDateTo.isDefined && bulkCurrentOutputInfoDate.isDefined) { - Some(BulkRunConfig(bulkCurrentDateFrom.get, bulkCurrentDateTo.get, infoDateColumn, bulkCurrentOutputInfoDate.get)) + Some(BulkRunConfig(bulkCurrentDateFrom.get, bulkCurrentDateTo.get, infoDateColumn, infoDateFormat2, bulkCurrentOutputInfoDate.get)) } else { None } diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/cmd/CmdLineConfig.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/cmd/CmdLineConfig.scala index dc42861a..34dd5bfa 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/cmd/CmdLineConfig.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/cmd/CmdLineConfig.scala @@ -45,6 +45,7 @@ case class CmdLineConfig( mode: Option[String] = None, bulkSize: Option[String] = None, infoDateColumn: Option[String] = None, + infoDateFormat: Option[String] = None, inverseOrder: Option[Boolean] = None, verbose: Option[Boolean] = None, overrideLogLevel: Option[String] = None, @@ -137,6 +138,9 @@ object CmdLineConfig { for (infoDateColumn <- cmd.infoDateColumn) accumulatedConfig = accumulatedConfig.withValue(INFO_DATE_COLUMN, ConfigValueFactory.fromAnyRef(infoDateColumn)) + for (infoDateColumn <- cmd.infoDateFormat) + accumulatedConfig = accumulatedConfig.withValue(INFO_DATE_FORMAT, ConfigValueFactory.fromAnyRef(infoDateColumn)) + for (logEffectiveConfig <- cmd.logEffectiveConfig) accumulatedConfig = accumulatedConfig.withValue(LOG_EFFECTIVE_CONFIG, ConfigValueFactory.fromAnyRef(logEffectiveConfig)) @@ -223,7 +227,13 @@ object CmdLineConfig { .text("The information date column name to use for repartitioning.") .validate(v => if (v.nonEmpty) success - else failure("Invalid information date column name. Must be a non-empty string.")) + else failure("Invalid information date column name. Must be a non-empty string.")), + opt[String]("info-date-format").optional().action((value, config) => + config.copy(infoDateFormat = Option(value))) + .text("The format of the information date column if it is not of date or datetime/timestamp type, for repartitioning.") + .validate(v => + if (v.nonEmpty) success + else failure("Invalid information date format. Must be a non-empty string.")) ) opt[Boolean]("inverse-order").optional().action((value, config) => diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistence.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistence.scala index 70694476..21be3a6f 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistence.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistence.scala @@ -43,9 +43,9 @@ trait MetastorePersistence { def isRepartitioningSupported: Boolean - def repartitionPhase1(infoDateColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {} + def repartitionPhase1(infoDateColumn: String, infoDateFormat: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {} - def repartitionPhase2(infoDateColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {} + def repartitionPhase2(infoDateColumn: String, infoDateFormat: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {} } object MetastorePersistence { diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceIceberg.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceIceberg.scala index aa27ab15..3cfb2bdd 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceIceberg.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceIceberg.scala @@ -18,7 +18,7 @@ package za.co.absa.pramen.core.metastore.peristence import org.apache.spark.sql._ import org.apache.spark.sql.functions._ -import org.apache.spark.sql.types.DateType +import org.apache.spark.sql.types.{DateType, StringType, TimestampType} import org.slf4j.LoggerFactory import za.co.absa.pramen.api.{CatalogTable, PartitionScheme} import za.co.absa.pramen.core.metastore.MetaTableStats @@ -113,25 +113,39 @@ class MetastorePersistenceIceberg(table: CatalogTable, throw new UnsupportedOperationException("Iceberg only operates on tables in a catalog. Separate Hive options are not supported.") } - override def isRepartitioningSupported: Boolean = true + override def isRepartitioningSupported: Boolean = partitionScheme != PartitionScheme.NotPartitioned - override def repartitionPhase1(infoDateDataColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = { + override def repartitionPhase1(infoDateDataColumn: String, infoDateDataFormat: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = { if (infoDateColumn.equalsIgnoreCase(infoDateDataColumn)) throw new IllegalArgumentException(s"Cannot repartition a table if the metastore info date column is the same as the data info date column ($infoDateDataColumn)") + if (partitionScheme == PartitionScheme.Overwrite) + throw new IllegalArgumentException(s"Repartitioning is not supported for this partition scheme: ${partitionScheme.getClass.getSimpleName}") + val fullTableName = table.getFullTableName val df = spark.table(fullTableName) .filter(getFilter(Some(outputInfoDate), Some(outputInfoDate))) + val dataInfoDateType = df.schema.fields + .find(_.name.equalsIgnoreCase(infoDateDataColumn)) + .map(_.dataType) + .getOrElse(StringType) + + val castExpression = dataInfoDateType match { + case _: DateType => col(infoDateDataColumn) + case _: TimestampType => col(infoDateDataColumn).cast(DateType) + case _ => to_date(col(infoDateDataColumn).cast(StringType), infoDateDataFormat) + } + log.info(s"Running Iceberg repartitioning: UPDATE $fullTableName SET $infoDateColumn = CAST($infoDateDataColumn AS DATE) " + s"WHERE $infoDateColumn = '$outputInfoDate' AND $infoDateDataColumn >= '$infoDateFrom' AND $infoDateDataColumn <= '$infoDateTo'") - val dfToWrite = df.withColumn(infoDateColumn, col(infoDateDataColumn).cast(DateType)) + val dfToWrite = df.withColumn(infoDateColumn, castExpression) writeRepartitionedDf(dfToWrite, fullTableName, infoDateColumn, infoDateFrom, infoDateTo, writeOptions) } - override def repartitionPhase2(infoDateDataColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = { + override def repartitionPhase2(infoDateDataColumn: String, infoDateDataFormat: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = { if (infoDateColumn.equalsIgnoreCase(infoDateDataColumn)) throw new IllegalArgumentException(s"Cannot repartition a table if the metastore info date column is the same as the data info date column ($infoDateDataColumn)") } diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceParquet.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceParquet.scala index aa5fab55..3e9648b3 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceParquet.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceParquet.scala @@ -18,6 +18,7 @@ package za.co.absa.pramen.core.metastore.peristence import org.apache.hadoop.fs.Path import org.apache.spark.sql.functions._ +import org.apache.spark.sql.types.{DateType, StringType, TimestampType} import org.apache.spark.sql.{Column, DataFrame, SaveMode, SparkSession} import org.slf4j.LoggerFactory import za.co.absa.pramen.api.{PartitionInfo, PartitionScheme} @@ -145,7 +146,86 @@ class MetastorePersistenceParquet(path: String, throw new UnsupportedOperationException("Parquet format does not support Hive tables at the moment.") } - override def isRepartitioningSupported: Boolean = false + override def isRepartitioningSupported: Boolean = partitionScheme == PartitionScheme.PartitionByDay + + override def repartitionPhase1(infoDateDataColumn: String, infoDateDataFormat: String, infoDateDataFrom: LocalDate, infoDateDataTo: LocalDate, outputInfoDate: LocalDate): Unit = { + ensureRepartitioningSupported(infoDateDataColumn) + + val infoDateFrom = outputInfoDate + val infoDateTo = outputInfoDate.plusYears(1000) + val pathFrom = SparkUtils.getPartitionPath(infoDateFrom, infoDateColumn, infoDateFormat, path) + val pathTo = SparkUtils.getPartitionPath(infoDateTo, infoDateColumn, infoDateFormat, path) + + val fsUtils = new FsUtils(spark.sparkContext.hadoopConfiguration, path) + + log.info(s"Repartitioning phase 1.1 - copying files to a temporary location ($pathTo)...") + fsUtils.copyDirectory(pathFrom, pathTo) + val sizeFrom = fsUtils.getDirectorySize(pathFrom.toUri.toString) + val sizeTo = fsUtils.getDirectorySize(pathTo.toUri.toString) + if (sizeFrom != sizeTo) { + throw new IllegalStateException(s"Repartitioning failed: size mismatch between $pathFrom ($sizeFrom) and $pathTo ($sizeTo)") + } + } + + override def repartitionPhase2(infoDateDataColumn: String, infoDateDataFormat: String, infoDateDataFrom: LocalDate, infoDateDataTo: LocalDate, outputInfoDate: LocalDate): Unit = { + ensureRepartitioningSupported(infoDateDataColumn) + + val infoDateFrom = outputInfoDate.plusYears(1000) + val pathFrom = SparkUtils.getPartitionPath(infoDateFrom, infoDateColumn, infoDateFormat, path) + val pathTo = SparkUtils.getPartitionPath(outputInfoDate, infoDateColumn, infoDateFormat, path) + val fsUtils = new FsUtils(spark.sparkContext.hadoopConfiguration, path) + if (!fsUtils.exists(pathFrom)) { + throw new IllegalArgumentException(s"Path does not exist: $pathFrom") + } + + log.info(s"Repartitioning phase 2.1 - deleting data in the original partition ($pathTo)...") + fsUtils.deleteDirectoryRecursively(pathTo) + + log.info(s"Repartitioning phase 2.2 - deleting data from target partitions ($infoDateDataFrom to $infoDateDataTo)...") + var date = infoDateDataFrom + while(date.isBefore(infoDateDataTo) || date.isEqual(infoDateDataTo)) { + val pathToDelete = SparkUtils.getPartitionPath(date, infoDateColumn, infoDateFormat, path) + fsUtils.deleteDirectoryRecursively(pathToDelete) + date = date.plusDays(1) + } + + log.info(s"Repartitioning phase 2.3 - appending data to target partitions ($infoDateDataFrom to $infoDateDataTo)...") + val df = spark.read + .format("parquet") + .options(readOptions) + .load(pathFrom.toUri.toString) + + val dataInfoDateType = df.schema.fields + .find(_.name.equalsIgnoreCase(infoDateDataColumn)) + .map(_.dataType) + .getOrElse(StringType) + + val castExpression = dataInfoDateType match { + case _: DateType => col(infoDateDataColumn) + case _: TimestampType => col(infoDateDataColumn).cast(DateType) + case _ => to_date(col(infoDateDataColumn).cast(StringType), infoDateDataFormat) + } + + df.withColumn(infoDateColumn, castExpression) + .write + .format("parquet") + .mode(SaveMode.Append) + .partitionBy(infoDateColumn) + .options(writeOptions) + .save(path) + + log.info(s"Repartitioning phase 2.4 - deleting data from the temporary location ($pathFrom)...") + fsUtils.deleteDirectoryRecursively(pathFrom) + } + + private def ensureRepartitioningSupported(infoDateDataColumn: String): Unit = { + if (infoDateColumn.equalsIgnoreCase(infoDateDataColumn)) + throw new IllegalArgumentException(s"Cannot repartition a table if the metastore info date column is the same as the data info date column ($infoDateDataColumn)") + + if (partitionScheme != PartitionScheme.PartitionByDay) + throw new IllegalArgumentException(s"Repartitioning is not supported for this partition scheme: ${partitionScheme.getClass.getSimpleName}") + } + def loadPartitionDirectly(infoDate: LocalDate): DataFrame = { val dateStr = dateFormatter.format(infoDate) diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/repartitioner/JobRepartitionerImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/repartitioner/JobRepartitionerImpl.scala index 8fa8eb4b..990354d0 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/repartitioner/JobRepartitionerImpl.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/repartitioner/JobRepartitionerImpl.scala @@ -85,7 +85,11 @@ class JobRepartitionerImpl(bulkLoadCurrent: BulkRunConfig, // Starting repartitioning phase 1 if (persistence.isRepartitioningSupported) { - persistence.repartitionPhase1(infoDateColumn, bulkLoadCurrent.outputInfoDate, bulkLoadCurrent.outputInfoDate, bulkLoadCurrent.outputInfoDate) + persistence.repartitionPhase1(infoDateColumn, + bulkLoadCurrent.infoDateFormat, + bulkLoadCurrent.dataDateFrom, + bulkLoadCurrent.dataDateTo, + bulkLoadCurrent.outputInfoDate) val updatedState = bulkLoadState.copy(phase = BulkLoadPhase.Repartition1) bulkLoadStateManager.updatePhase(updatedState) @@ -101,7 +105,11 @@ class JobRepartitionerImpl(bulkLoadCurrent: BulkRunConfig, // Starting repartitioning phase 2 if (persistence.isRepartitioningSupported) { - persistence.repartitionPhase2(infoDateColumn, bulkLoadCurrent.outputInfoDate, bulkLoadCurrent.outputInfoDate, bulkLoadCurrent.outputInfoDate) + persistence.repartitionPhase2(infoDateColumn, + bulkLoadCurrent.infoDateFormat, + bulkLoadCurrent.dataDateFrom, + bulkLoadCurrent.dataDateTo, + bulkLoadCurrent.outputInfoDate) val updatedState = bulkLoadState.copy(phase = BulkLoadPhase.Done) bulkLoadStateManager.updatePhase(updatedState) BulkLoadPhase.Done diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/FsUtils.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/FsUtils.scala index 1b3bc428..bcb8faaa 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/FsUtils.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/FsUtils.scala @@ -466,6 +466,38 @@ class FsUtils(conf: Configuration, pathBase: String) { } } + /** + * Copies the contents of a directory to another location. + * + * The source directory must exist, otherwise an `IllegalArgumentException` is thrown. + * If the destination directory already exists, it is deleted recursively and + * recreated before the copy operation starts. Every file found directly in the + * source directory is then copied into the destination directory, keeping its + * original name, using a retrying copy operation to tolerate transient failures. + * + * @param srcDir the path of the existing directory whose files are copied + * @param dstDir the path of the target directory, recreated from scratch before copying + * @return nothing, the method is executed only for its side effects on the file system + * @throws IllegalArgumentException if the source path does not exist + */ + def copyDirectory(srcDir: Path, dstDir: Path): Unit = { + if (!exists(srcDir)) + throw new IllegalArgumentException("No data in the source path: " + srcDir) + + if (exists(dstDir)) + deleteDirectoryRecursively(dstDir) + + createDirectoryRecursive(dstDir) + + val listOfFiles = getHadoopFiles(srcDir, includeHiddenFiles = true) + + listOfFiles.foreach(file => { + val srcPath = file.getPath + val dstFile = new Path(dstDir, srcPath.getName) + copyFileWithRetry(srcPath, dstFile) + }) + } + def copyToLocal(srcFile: Path, targetFile: Path, overwrite: Boolean = false): Unit = { if (!overwrite && fs.exists(targetFile)) { throw new IllegalStateException(s"Target file $targetFile already exists.") diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/cmd/CmdLineConfigSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/cmd/CmdLineConfigSuite.scala index 702963ec..9c5995b8 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/cmd/CmdLineConfigSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/cmd/CmdLineConfigSuite.scala @@ -244,7 +244,7 @@ class CmdLineConfigSuite extends AnyWordSpec { } "return a modified config if bulk mode for date-to override is specified" in { - val cmd = CmdLineConfig.parseCmdLine(Array("--workflow", "dummy.config", "--date-to", "2020-08-15", "--inverse-order", "true", "--run-mode", "bulk", "--bulk-size", "yearly", "--info-date-column", "info_date")) + val cmd = CmdLineConfig.parseCmdLine(Array("--workflow", "dummy.config", "--date-to", "2020-08-15", "--inverse-order", "true", "--run-mode", "bulk", "--bulk-size", "yearly", "--info-date-column", "info_date", "--info-date-format", "yyyyMMdd")) val config = CmdLineConfig.applyCmdLineToConfig(emptyConfig, cmd.get) assert(config.hasPath(LOAD_DATE_TO)) @@ -256,6 +256,7 @@ class CmdLineConfigSuite extends AnyWordSpec { assert(config.getString(RUN_MODE) == "bulk") assert(config.getString(RUN_BULK_BATCH_SIZE) == "yearly") assert(config.getString(INFO_DATE_COLUMN) == "info_date") + assert(config.getString(INFO_DATE_FORMAT) == "yyyyMMdd") } "return the original config if no cmd line arguments are provided" in { diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/integration/BulkLoadLongSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/integration/BulkLoadLongSuite.scala index 1516374d..e46c7706 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/integration/BulkLoadLongSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/integration/BulkLoadLongSuite.scala @@ -154,16 +154,16 @@ class BulkLoadLongSuite extends AnyWordSpec assert(df.count() == 59) - // For now... - assert(!df.filter(col("dt") =!= col("pramen_info_date")).isEmpty) + // Ensure parquet directory was repartitioned + assert(df.filter(!(col("dt") <=> col("pramen_info_date"))).isEmpty) - // Running the job for the second time shouyld not change the output + // Running the job for the second time should not change the output val exitCode2 = AppRunner.runBulkPipelines(conf) assert(exitCode2 == 0) val df2 = spark.table(tableName) assert(df2.count() == 59) - assert(df2.filter(col("dt") =!= col("pramen_info_date")).isEmpty) + assert(df2.filter(!(col("dt") <=> col("pramen_info_date"))).isEmpty) spark.sql(s"DELETE FROM $tableName").count() }