From 5e4c6f4627aa4124b0592c63c6a014b7f045e24d Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Thu, 8 Oct 2026 09:02:22 +0200 Subject: [PATCH 1/5] Add `Cobrix.validateOptions()` builder for validating spark-cobol options against data files without loading them into a DataFrame. Also rename `HDFSUtils` to `HadoopFsUtils` and make `DefaultSource.isRecursiveRetrieval` public. --- .../co/absa/cobrix/spark/cobol/Cobrix.scala | 25 +++++++- .../cobol/builder/SparkCobolBuilder.scala | 10 +-- .../cobol/builder/ValidatorBuilder.scala | 64 +++++++++++++++++++ .../spark/cobol/source/DefaultSource.scala | 16 ++--- .../copybook/CopybookContentLoader.scala | 5 +- .../cobol/source/index/IndexBuilder.scala | 6 +- .../source/streaming/CobolStreamer.scala | 4 +- .../{HDFSUtils.scala => HadoopFsUtils.scala} | 4 +- .../cobrix/spark/cobol/utils/SparkUtils.scala | 2 +- .../Test1FixedLengthRecordsSpec.scala | 23 ++++++- ...tilsSpec.scala => HadoopFsUtilsSpec.scala} | 19 +++--- 11 files changed, 141 insertions(+), 37 deletions(-) create mode 100644 spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala rename spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/{HDFSUtils.scala => HadoopFsUtils.scala} (96%) rename spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/utils/{HDFSUtilsSpec.scala => HadoopFsUtilsSpec.scala} (82%) diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/Cobrix.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/Cobrix.scala index 26837e160..4dddb7840 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/Cobrix.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/Cobrix.scala @@ -17,10 +17,33 @@ package za.co.absa.cobrix.spark.cobol import org.apache.spark.sql.SparkSession -import za.co.absa.cobrix.spark.cobol.builder.SparkCobolBuilder +import za.co.absa.cobrix.spark.cobol.builder.{SparkCobolBuilder, ValidatorBuilder} object Cobrix { + /** + * Creates a builder for converting an RDD of COBOL records into a Spark DataFrame. + * + * The returned builder is used to specify the copybook that describes the record layout, + * either by providing its contents directly or by giving a path to a copybook file. + * Further parsing options can then be configured before the conversion is performed. + * + * @return a new [[SparkCobolBuilder]] instance for specifying the copybook and conversion options + */ def fromRdd(implicit spark: SparkSession): SparkCobolBuilder = { new SparkCobolBuilder() } + + /** + * Creates a builder for validating spark-cobol options against a set of data files. + * + * The returned builder is used to specify reader options, either one at a time or as a map. + * Option keys are treated case-insensitively. The data files can then be validated against + * these options without fully loading them into a DataFrame. File paths passed for validation + * may contain wildcards. + * + * @return a new [[ValidatorBuilder]] instance for specifying options and validating data files + */ + def validateOptions(implicit spark: SparkSession): ValidatorBuilder = { + new ValidatorBuilder() + } } diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/SparkCobolBuilder.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/SparkCobolBuilder.scala index 11c764828..105bd02c0 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/SparkCobolBuilder.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/SparkCobolBuilder.scala @@ -17,15 +17,15 @@ package za.co.absa.cobrix.spark.cobol.builder import org.apache.spark.sql.SparkSession -import za.co.absa.cobrix.spark.cobol.utils.HDFSUtils +import za.co.absa.cobrix.spark.cobol.utils.HadoopFsUtils -class SparkCobolBuilder { - def copybookContents(copybookContent: String)(implicit spark: SparkSession): SparkCobolOptionsBuilder = { +class SparkCobolBuilder(implicit spark: SparkSession) { + def copybookContents(copybookContent: String): SparkCobolOptionsBuilder = { new SparkCobolOptionsBuilder(copybookContent) } - def copybookPath(copybookPath: String)(implicit spark: SparkSession): SparkCobolOptionsBuilder = { - val copybook = HDFSUtils.loadTextFileFromHadoop(spark.sparkContext.hadoopConfiguration, copybookPath) + def copybookPath(copybookPath: String): SparkCobolOptionsBuilder = { + val copybook = HadoopFsUtils.loadTextFileFromHadoop(spark.sparkContext.hadoopConfiguration, copybookPath) new SparkCobolOptionsBuilder(copybook) } } diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala new file mode 100644 index 000000000..71b4d6121 --- /dev/null +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala @@ -0,0 +1,64 @@ +/* + * Copyright 2018 ABSA Group Limited + * + * 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 za.co.absa.cobrix.spark.cobol.builder + +import org.apache.spark.sql.SparkSession +import za.co.absa.cobrix.cobol.reader.parameters.{CobolParameters, CobolParametersParser, Parameters} +import za.co.absa.cobrix.spark.cobol.source.DefaultSource.buildEitherReader +import za.co.absa.cobrix.spark.cobol.source.parameters.LocalityParameters +import za.co.absa.cobrix.spark.cobol.source.{CobolRelation, DefaultSource} + +import scala.collection.mutable + +class ValidatorBuilder(implicit spark: SparkSession) { + private val caseInsensitiveOptions = new mutable.HashMap[String, String]() + + def option(key: String, value: String): ValidatorBuilder = { + caseInsensitiveOptions += (key.toLowerCase -> value) + this + } + + def options(options: Map[String, String]): ValidatorBuilder = { + caseInsensitiveOptions ++= options.map(kv => (kv._1.toLowerCase(), kv._2)) + this + } + + + /** + * Validates the list of data files against spark-cobol options. + * File paths can contain wildcards, e.g. "/somedur/test*" + */ + def validate(filePaths: String*): Unit = { + val sqlContext = spark.sqlContext + + val cobolParameters: CobolParameters = CobolParametersParser.parse(new Parameters(caseInsensitiveOptions.toMap)) + .copy(sourcePaths = filePaths) + val isRecursiveRetrieval = DefaultSource.isRecursiveRetrieval(sqlContext) + val filesList = CobolRelation.getListFilesWithOrder(filePaths, sqlContext, isRecursiveRetrieval) + val hasGpg = cobolParameters.gpgPrivateKey.isDefined + val hasCompressedFiles = hasGpg || filesList.exists(_.isCompressed) + + val relation = new CobolRelation(cobolParameters.sourcePaths, + filesList, + buildEitherReader(sqlContext.sparkSession, cobolParameters, hasCompressedFiles), + LocalityParameters.extract(cobolParameters), + cobolParameters.debugIgnoreFileSize, + cobolParameters.recordLimit)(sqlContext) + + relation.buildScan() + } +} diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/DefaultSource.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/DefaultSource.scala index 7a741b055..310c4c46f 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/DefaultSource.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/DefaultSource.scala @@ -141,14 +141,6 @@ class DefaultSource //TODO fix with the correct implementation once the correct Reader hierarchy is put in place. override def buildReader(spark: SparkSession, parameters: Map[String, String]): FixedLenReader = null - - /** - * Checks if the recursive file retrieval flag is set - */ - private def isRecursiveRetrieval(sqlContext: SQLContext): Boolean = { - val hadoopConf = sqlContext.sparkContext.hadoopConfiguration - hadoopConf.getBoolean(FileInputFormat.INPUT_DIR_RECURSIVE, false) - } } object DefaultSource { @@ -322,4 +314,12 @@ object DefaultSource { copybookContent, getReaderProperties(parameters, defaultHdfsBlockSize) ) } + + /** + * Checks if the recursive file retrieval flag is set + */ + def isRecursiveRetrieval(sqlContext: SQLContext): Boolean = { + val hadoopConf = sqlContext.sparkContext.hadoopConfiguration + hadoopConf.getBoolean(FileInputFormat.INPUT_DIR_RECURSIVE, false) + } } diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/copybook/CopybookContentLoader.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/copybook/CopybookContentLoader.scala index 3453413bf..d902765ad 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/copybook/CopybookContentLoader.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/copybook/CopybookContentLoader.scala @@ -18,8 +18,7 @@ package za.co.absa.cobrix.spark.cobol.source.copybook import org.apache.hadoop.conf.Configuration import za.co.absa.cobrix.cobol.reader.parameters.CobolParameters -import za.co.absa.cobrix.spark.cobol.utils.FsType.LocalFs -import za.co.absa.cobrix.spark.cobol.utils.{FileNameUtils, FsType, HDFSUtils, ResourceUtils} +import za.co.absa.cobrix.spark.cobol.utils.{FileNameUtils, FsType, HadoopFsUtils, ResourceUtils} import java.nio.charset.StandardCharsets import java.nio.file.{Files, Paths} @@ -46,7 +45,7 @@ object CopybookContentLoader { fsType match { case FsType.LocalFs => loadCopybookFromLocalFS(copyBookFileName) case FsType.JarFs => loadCopybookFromJarResources(copyBookFileName) - case FsType.HadoopFs => HDFSUtils.loadTextFileFromHadoop(hadoopConf, copyBookFileName) + case FsType.HadoopFs => HadoopFsUtils.loadTextFileFromHadoop(hadoopConf, copyBookFileName) } } diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/index/IndexBuilder.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/index/IndexBuilder.scala index 8e724fbfd..7e0d8301e 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/index/IndexBuilder.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/index/IndexBuilder.scala @@ -32,7 +32,7 @@ import za.co.absa.cobrix.spark.cobol.source.SerializableConfiguration import za.co.absa.cobrix.spark.cobol.source.parameters.LocalityParameters import za.co.absa.cobrix.spark.cobol.source.streaming.FileStreamer import za.co.absa.cobrix.spark.cobol.source.types.FileWithOrder -import za.co.absa.cobrix.spark.cobol.utils.{FileUtils, HDFSUtils, LRUCache, SparkUtils} +import za.co.absa.cobrix.spark.cobol.utils.{FileUtils, HadoopFsUtils, LRUCache, SparkUtils} import scala.collection.mutable.ArrayBuffer @@ -91,7 +91,7 @@ private[cobol] object IndexBuilder extends Logging { index.map { entry => val offset = if (entry.offsetFrom >= 0) entry.offsetFrom else 0 val length = getBlockLengthByIndexEntry(entry) - (entry, HDFSUtils.getBlocksLocations(new Path(filePath), offset, length, fileSystem)) + (entry, HadoopFsUtils.getBlocksLocations(new Path(filePath), offset, length, fileSystem)) } } } @@ -359,7 +359,7 @@ private[cobol] object IndexBuilder extends Logging { val fileSystem = FileSystem.get(conf) val filesWithPreferredLocations = filesList.map(file => { - (file, HDFSUtils.getBlocksLocations(new Path(file.filePath), fileSystem)) + (file, HadoopFsUtils.getBlocksLocations(new Path(file.filePath), fileSystem)) }).toSeq filesWithPreferredLocations.foreach(a => logDebug(a.toString())) diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/CobolStreamer.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/CobolStreamer.scala index ff452639a..144cd1d66 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/CobolStreamer.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/CobolStreamer.scala @@ -24,7 +24,7 @@ import za.co.absa.cobrix.cobol.reader.parameters.CobolParametersParser._ import za.co.absa.cobrix.cobol.reader.parameters.ReaderParameters import za.co.absa.cobrix.spark.cobol.reader.{FixedLenNestedReader, FixedLenReader} import za.co.absa.cobrix.spark.cobol.source.parameters.CobolParametersValidator -import za.co.absa.cobrix.spark.cobol.utils.HDFSUtils +import za.co.absa.cobrix.spark.cobol.utils.HadoopFsUtils /** * Provides an integration point for adding streaming support to the Spark-Cobol library. @@ -34,7 +34,7 @@ import za.co.absa.cobrix.spark.cobol.utils.HDFSUtils object CobolStreamer { def getReader(implicit ssc: StreamingContext): FixedLenReader = { - val copybooks = Seq(HDFSUtils.loadTextFileFromHadoop(ssc.sparkContext.hadoopConfiguration, ssc.sparkContext.getConf.get(PARAM_COPYBOOK_PATH))) + val copybooks = Seq(HadoopFsUtils.loadTextFileFromHadoop(ssc.sparkContext.hadoopConfiguration, ssc.sparkContext.getConf.get(PARAM_COPYBOOK_PATH))) new FixedLenNestedReader(copybooks, readerProperties = ReaderParameters()) } diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/HDFSUtils.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/HadoopFsUtils.scala similarity index 96% rename from spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/HDFSUtils.scala rename to spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/HadoopFsUtils.scala index 4932e3db5..2b7f35210 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/HDFSUtils.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/HadoopFsUtils.scala @@ -26,9 +26,9 @@ import scala.collection.JavaConverters._ import scala.util.control.NonFatal /** - * This object provides utility methods for interacting with HDFS internals. + * This object provides utility methods for interacting with Hadoop filesystem internals. */ -object HDFSUtils { +object HadoopFsUtils { private val log: Logger = LoggerFactory.getLogger(this.getClass) final val bytesInMegabyte = 1048576 diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/SparkUtils.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/SparkUtils.scala index 744b8afc7..65c9741db 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/SparkUtils.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/SparkUtils.scala @@ -512,7 +512,7 @@ object SparkUtils extends Logging { case None => FileSystem.get(conf) } - val hdfsBlockSize = HDFSUtils.getHDFSDefaultBlockSizeMB(fileSystem) + val hdfsBlockSize = HadoopFsUtils.getHDFSDefaultBlockSizeMB(fileSystem) hdfsBlockSize match { case None => logger.info(s"Unable to get default block size for '${fileSystem.getScheme}://.") case Some(size) => logger.info(s"Default block size for '${fileSystem.getScheme}://' is $size MB.") diff --git a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test1FixedLengthRecordsSpec.scala b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test1FixedLengthRecordsSpec.scala index b2b2f0e28..cc5217399 100644 --- a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test1FixedLengthRecordsSpec.scala +++ b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test1FixedLengthRecordsSpec.scala @@ -16,13 +16,14 @@ package za.co.absa.cobrix.spark.cobol.source.integration -import java.nio.charset.StandardCharsets -import java.nio.file.{Files, Paths} - import org.scalatest.funsuite.AnyFunSuite +import za.co.absa.cobrix.spark.cobol.Cobrix import za.co.absa.cobrix.spark.cobol.source.base.SparkTestBase import za.co.absa.cobrix.spark.cobol.utils.{FileUtils, SparkUtils} +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Paths} + //noinspection NameBooleanParameters class Test1FixedLengthRecordsSpec extends AnyFunSuite with SparkTestBase { @@ -87,6 +88,22 @@ class Test1FixedLengthRecordsSpec extends AnyFunSuite with SparkTestBase { assert(exception.getMessage.contains("NOT DIVISIBLE by the RECORD SIZE")) } + test(s"Test failure of validation on an invalid copybook") { + val copybook = + """ 01 COMPANY-DETAILS. + | 05 SEGMENT-ID PIC X(5). + | 05 COMPANY-ID PIC X(11). + |""".stripMargin + + val exception = intercept[IllegalArgumentException]{ + Cobrix.validateOptions + .option("copybook_contents", copybook) + .option("schema_retention_policy", "collapse_root") + .validate(inpudDataPath) + } + assert(exception.getMessage.contains("NOT DIVISIBLE by the RECORD SIZE")) + } + test(s"Test success on an invalid copybook with debug override") { val copybook = """ 01 COMPANY-DETAILS. diff --git a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/utils/HDFSUtilsSpec.scala b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/utils/HadoopFsUtilsSpec.scala similarity index 82% rename from spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/utils/HDFSUtilsSpec.scala rename to spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/utils/HadoopFsUtilsSpec.scala index ece3fdf6b..b87c35bd6 100644 --- a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/utils/HDFSUtilsSpec.scala +++ b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/utils/HadoopFsUtilsSpec.scala @@ -16,15 +16,16 @@ package za.co.absa.cobrix.spark.cobol.utils -import java.io.{DataOutputStream, File, FileOutputStream} -import java.util.UUID -import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} import org.apache.commons.io.{FileUtils => CommonsFileUtils} import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.{FileSystem, Path} import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} + +import java.io.{DataOutputStream, File, FileOutputStream} +import java.util.UUID -class HDFSUtilsSpec extends AnyFlatSpec with BeforeAndAfterAll with BeforeAndAfterEach { +class HadoopFsUtilsSpec extends AnyFlatSpec with BeforeAndAfterAll with BeforeAndAfterEach { private val baseTestDir = TempDir.getNew private val validFile = new File(baseTestDir, "a_valid_file") @@ -42,17 +43,17 @@ class HDFSUtilsSpec extends AnyFlatSpec with BeforeAndAfterAll with BeforeAndAft CommonsFileUtils.deleteDirectory(baseTestDir) } - behavior of HDFSUtils.getClass.getName + behavior of HadoopFsUtils.getClass.getName it should "throw if offsets are invalid" in { val errorLength = intercept[IllegalArgumentException] { - HDFSUtils.getBlocksLocations(toHDFSPath(validFile), 0, 0, fileSystem) + HadoopFsUtils.getBlocksLocations(toHDFSPath(validFile), 0, 0, fileSystem) } assert(errorLength.getMessage.contains("Invalid")) val errorOffset = intercept[IllegalArgumentException] { - HDFSUtils.getBlocksLocations(toHDFSPath(validFile), -1, 10, fileSystem) + HadoopFsUtils.getBlocksLocations(toHDFSPath(validFile), -1, 10, fileSystem) } assert(errorOffset.getMessage.contains("Invalid")) } @@ -60,14 +61,14 @@ class HDFSUtilsSpec extends AnyFlatSpec with BeforeAndAfterAll with BeforeAndAft it should "throw if file is actually a directory" in { val error = intercept[IllegalArgumentException] { - HDFSUtils.getBlocksLocations(toHDFSPath(invalidFile), 0, 10, fileSystem) + HadoopFsUtils.getBlocksLocations(toHDFSPath(invalidFile), 0, 10, fileSystem) } assert(!error.getMessage.isEmpty) } it should "return the correct locations for the files blocks" in { - val locations = HDFSUtils.getBlocksLocations(toHDFSPath(validFile), 0, 10, fileSystem) + val locations = HadoopFsUtils.getBlocksLocations(toHDFSPath(validFile), 0, 10, fileSystem) assert(locations.size == 1) assert(locations.head == "localhost") From dba7421ddcfb71a42f695fa596a4dbcaf82f74e5 Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Thu, 8 Oct 2026 09:54:50 +0200 Subject: [PATCH 2/5] Fix 'spark-cobol' options validator to never look inside files, only check option consistency and that file sizes are acceptable. --- README.md | 14 ++++++++ .../cobol/builder/ValidatorBuilder.scala | 8 +++-- .../spark/cobol/source/CobolRelation.scala | 8 +++++ .../cobol/source/scanners/CobolScanners.scala | 35 ++++++++++--------- 4 files changed, 46 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index 33abccd2a..02b141460 100644 --- a/README.md +++ b/README.md @@ -829,6 +829,20 @@ val df = Cobrix.fromRdd When converting from an RDD some of the options like `record_format` or `generate_record_id` cannot be used since the data is assumed to be already split by records and the information about file names and relative order of records is not available. +### Options validation +You can validate options to `spark-cobol` options without reading data files. This is useful for quick validation +separately from the routine that processes the files. + +An example: +```scala +import za.co.absa.cobrix.spark.cobol.Cobrix + +Cobrix.validateOptions + .option("copybook_contents", copybook) + .option("some_option", "some_value") + .validate(inpudDataPath) +``` + ## EBCDIC code pages The following code pages are supported: diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala index 71b4d6121..ce00ffd83 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala @@ -45,8 +45,10 @@ class ValidatorBuilder(implicit spark: SparkSession) { def validate(filePaths: String*): Unit = { val sqlContext = spark.sqlContext - val cobolParameters: CobolParameters = CobolParametersParser.parse(new Parameters(caseInsensitiveOptions.toMap)) - .copy(sourcePaths = filePaths) + val cobolParameters0: CobolParameters = CobolParametersParser.parse(new Parameters(caseInsensitiveOptions.toMap)) + val varLenOptions = cobolParameters0.variableLengthParams.map(_.copy(generateRecordId = false)) + val cobolParameters: CobolParameters = cobolParameters0 + .copy(sourcePaths = filePaths, generateRecordBytes = false, debugIgnoreFileSize = false, variableLengthParams = varLenOptions) val isRecursiveRetrieval = DefaultSource.isRecursiveRetrieval(sqlContext) val filesList = CobolRelation.getListFilesWithOrder(filePaths, sqlContext, isRecursiveRetrieval) val hasGpg = cobolParameters.gpgPrivateKey.isDefined @@ -59,6 +61,6 @@ class ValidatorBuilder(implicit spark: SparkSession) { cobolParameters.debugIgnoreFileSize, cobolParameters.recordLimit)(sqlContext) - relation.buildScan() + relation.validateRelation() } } diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/CobolRelation.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/CobolRelation.scala index 9b3c8fcd0..a35737eb5 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/CobolRelation.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/CobolRelation.scala @@ -104,6 +104,14 @@ class CobolRelation(sourceDirs: Seq[String], } }) } + + private[cobrix] def validateRelation(): Unit = { + cobolReader match { + case blockReader: FixedLenReader => + CobolScanners.buildScanForFixedLength(blockReader, sourceDirs, parseRecords, debugIgnoreFileSize, sqlContext) + case _ => // skip for other readers + } + } } object CobolRelation { diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/scanners/CobolScanners.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/scanners/CobolScanners.scala index 33522e561..88908b6cc 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/scanners/CobolScanners.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/scanners/CobolScanners.scala @@ -99,28 +99,31 @@ private[source] object CobolScanners extends Logging { val conf = new Configuration(sqlContext.sparkContext.hadoopConfiguration) val recordSize = reader.getRecordSize - sourceDirs.foreach(sourceDir => { - if (!debugIgnoreFileSize) { - val nonDivisibleFiles = getNonDivisibleFiles(sourceDir, conf, recordSize) - - if (nonDivisibleFiles.nonEmpty) { - nonDivisibleFiles.head match { - case (name, size) => - if (nonDivisibleFiles.length > 1) { - throw new IllegalArgumentException(s"Multiple file sizes are NOT DIVISIBLE by the RECORD SIZE calculated from the copybook ($recordSize bytes per record). Example file: $name size ($size bytes).") - } else { - throw new IllegalArgumentException(s"File $name size ($size bytes) is NOT DIVISIBLE by the RECORD SIZE calculated from the copybook ($recordSize bytes per record).") - } - } - } - } - }) + if (!debugIgnoreFileSize) + validateFixedRecordLengthFileSizes(sourceDirs, conf, recordSize) val records = sourceDirs.map(sourceDir => sqlContext.sparkContext.binaryRecords(sourceDir, recordSize, conf)) .reduce((a ,b) => a.union(b)) recordParser(reader, records) } + private[source] def validateFixedRecordLengthFileSizes(sourceDirs: Seq[String], conf: Configuration, recordSize: Int): Unit = { + sourceDirs.foreach(sourceDir => { + val nonDivisibleFiles = getNonDivisibleFiles(sourceDir, conf, recordSize) + + if (nonDivisibleFiles.nonEmpty) { + nonDivisibleFiles.head match { + case (name, size) => + if (nonDivisibleFiles.length > 1) { + throw new IllegalArgumentException(s"Multiple file sizes are NOT DIVISIBLE by the RECORD SIZE calculated from the copybook ($recordSize bytes per record). Example file: $name size ($size bytes).") + } else { + throw new IllegalArgumentException(s"File $name size ($size bytes) is NOT DIVISIBLE by the RECORD SIZE calculated from the copybook ($recordSize bytes per record).") + } + } + } + }) + } + private[source] def buildScanForTextFiles(reader: FixedLenTextReader, sourceDirs: Seq[String], recordParser: (FixedLenTextReader, RDD[Array[Byte]]) => RDD[Row], From 90c7513e9ff45d7478c53301493b9577df422176 Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Thu, 8 Oct 2026 10:39:50 +0200 Subject: [PATCH 3/5] Skip fixed-record length size divisibility check for text files since text files do not require that. --- .../za/co/absa/cobrix/spark/cobol/source/CobolRelation.scala | 1 + 1 file changed, 1 insertion(+) diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/CobolRelation.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/CobolRelation.scala index a35737eb5..bf6c9f418 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/CobolRelation.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/CobolRelation.scala @@ -107,6 +107,7 @@ class CobolRelation(sourceDirs: Seq[String], private[cobrix] def validateRelation(): Unit = { cobolReader match { + case _: FixedLenTextReader => // skip for text files. Even though FixedLenTextReader extends FixedLenReader, it can handle size mismatches case blockReader: FixedLenReader => CobolScanners.buildScanForFixedLength(blockReader, sourceDirs, parseRecords, debugIgnoreFileSize, sqlContext) case _ => // skip for other readers From 6ec5f9e2ae2e6d86ba3d0813644c9a0e825aaaa2 Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Thu, 8 Oct 2026 11:19:29 +0200 Subject: [PATCH 4/5] Rename validate() to validateLoad() since this validation works only when reading data. validatwSave() will be implemented later. --- README.md | 2 +- .../spark/cobol/builder/ValidatorBuilder.scala | 16 ++++++++++++++-- .../Test1FixedLengthRecordsSpec.scala | 2 +- 3 files changed, 16 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 02b141460..9743795b1 100644 --- a/README.md +++ b/README.md @@ -840,7 +840,7 @@ import za.co.absa.cobrix.spark.cobol.Cobrix Cobrix.validateOptions .option("copybook_contents", copybook) .option("some_option", "some_value") - .validate(inpudDataPath) + .validateLoad(inpudDataPath) ``` ## EBCDIC code pages diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala index ce00ffd83..90662af65 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala @@ -17,9 +17,10 @@ package za.co.absa.cobrix.spark.cobol.builder import org.apache.spark.sql.SparkSession +import za.co.absa.cobrix.cobol.reader.parameters.CobolParametersParser.PARAM_SOURCE_PATHS import za.co.absa.cobrix.cobol.reader.parameters.{CobolParameters, CobolParametersParser, Parameters} import za.co.absa.cobrix.spark.cobol.source.DefaultSource.buildEitherReader -import za.co.absa.cobrix.spark.cobol.source.parameters.LocalityParameters +import za.co.absa.cobrix.spark.cobol.source.parameters.{CobolParametersValidator, LocalityParameters} import za.co.absa.cobrix.spark.cobol.source.{CobolRelation, DefaultSource} import scala.collection.mutable @@ -39,16 +40,26 @@ class ValidatorBuilder(implicit spark: SparkSession) { /** + * This methods validates options when reading data files. + * * Validates the list of data files against spark-cobol options. * File paths can contain wildcards, e.g. "/somedur/test*" */ - def validate(filePaths: String*): Unit = { + def validateLoad(filePaths: String*): Unit = { val sqlContext = spark.sqlContext + // Validating options compatibility + val cobolOptions = caseInsensitiveOptions.toMap + (PARAM_SOURCE_PATHS -> filePaths.mkString(",")) + CobolParametersValidator.validateOrThrow(cobolOptions, sqlContext.sparkSession.sparkContext.hadoopConfiguration) + val cobolParameters0: CobolParameters = CobolParametersParser.parse(new Parameters(caseInsensitiveOptions.toMap)) val varLenOptions = cobolParameters0.variableLengthParams.map(_.copy(generateRecordId = false)) val cobolParameters: CobolParameters = cobolParameters0 .copy(sourcePaths = filePaths, generateRecordBytes = false, debugIgnoreFileSize = false, variableLengthParams = varLenOptions) + + // Validating cobol parameters after parsing + CobolParametersValidator.checkSanity(cobolParameters) + val isRecursiveRetrieval = DefaultSource.isRecursiveRetrieval(sqlContext) val filesList = CobolRelation.getListFilesWithOrder(filePaths, sqlContext, isRecursiveRetrieval) val hasGpg = cobolParameters.gpgPrivateKey.isDefined @@ -61,6 +72,7 @@ class ValidatorBuilder(implicit spark: SparkSession) { cobolParameters.debugIgnoreFileSize, cobolParameters.recordLimit)(sqlContext) + // Validating reader requirements relation.validateRelation() } } diff --git a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test1FixedLengthRecordsSpec.scala b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test1FixedLengthRecordsSpec.scala index cc5217399..4a93a025d 100644 --- a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test1FixedLengthRecordsSpec.scala +++ b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test1FixedLengthRecordsSpec.scala @@ -99,7 +99,7 @@ class Test1FixedLengthRecordsSpec extends AnyFunSuite with SparkTestBase { Cobrix.validateOptions .option("copybook_contents", copybook) .option("schema_retention_policy", "collapse_root") - .validate(inpudDataPath) + .validateLoad(inpudDataPath) } assert(exception.getMessage.contains("NOT DIVISIBLE by the RECORD SIZE")) } From 853cf620ba3cb3a6ef146116fc50ef145a731ed1 Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Fri, 9 Oct 2026 08:14:20 +0200 Subject: [PATCH 5/5] Add validator for options used to write data files from a dataframe. --- README.md | 8 +++++- .../cobol/builder/ValidatorBuilder.scala | 27 +++++++++++++++++-- .../writer/FixedLengthEbcdicWriterSuite.scala | 21 +++++++++++++++ 3 files changed, 53 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 9743795b1..63c0ba958 100644 --- a/README.md +++ b/README.md @@ -830,17 +830,23 @@ When converting from an RDD some of the options like `record_format` or `generat data is assumed to be already split by records and the information about file names and relative order of records is not available. ### Options validation -You can validate options to `spark-cobol` options without reading data files. This is useful for quick validation +You can validate options to `spark-cobol` options without reading ro writing data files. This is useful for quick validation separately from the routine that processes the files. An example: ```scala import za.co.absa.cobrix.spark.cobol.Cobrix +// Validation options when reading data Cobrix.validateOptions .option("copybook_contents", copybook) .option("some_option", "some_value") .validateLoad(inpudDataPath) + +// Validation options when writing data +Cobrix.validateOptions + .option("copybook_contents", copybookContents) + .validateSave(df) ``` ## EBCDIC code pages diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala index 90662af65..d66b4398c 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala @@ -16,12 +16,15 @@ package za.co.absa.cobrix.spark.cobol.builder -import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.{DataFrame, SparkSession} import za.co.absa.cobrix.cobol.reader.parameters.CobolParametersParser.PARAM_SOURCE_PATHS import za.co.absa.cobrix.cobol.reader.parameters.{CobolParameters, CobolParametersParser, Parameters} +import za.co.absa.cobrix.cobol.reader.schema.CobolSchema import za.co.absa.cobrix.spark.cobol.source.DefaultSource.buildEitherReader +import za.co.absa.cobrix.spark.cobol.source.copybook.CopybookContentLoader import za.co.absa.cobrix.spark.cobol.source.parameters.{CobolParametersValidator, LocalityParameters} import za.co.absa.cobrix.spark.cobol.source.{CobolRelation, DefaultSource} +import za.co.absa.cobrix.spark.cobol.writer.RecordCombinerSelector import scala.collection.mutable @@ -50,7 +53,7 @@ class ValidatorBuilder(implicit spark: SparkSession) { // Validating options compatibility val cobolOptions = caseInsensitiveOptions.toMap + (PARAM_SOURCE_PATHS -> filePaths.mkString(",")) - CobolParametersValidator.validateOrThrow(cobolOptions, sqlContext.sparkSession.sparkContext.hadoopConfiguration) + CobolParametersValidator.validateOrThrow(cobolOptions, spark.sparkContext.hadoopConfiguration) val cobolParameters0: CobolParameters = CobolParametersParser.parse(new Parameters(caseInsensitiveOptions.toMap)) val varLenOptions = cobolParameters0.variableLengthParams.map(_.copy(generateRecordId = false)) @@ -75,4 +78,24 @@ class ValidatorBuilder(implicit spark: SparkSession) { // Validating reader requirements relation.validateRelation() } + + /** + * This method validates options when writing a DataFrame. + * + * Validates spark-cobol options are consistent with a write request. + */ + def validateSave(df: DataFrame): Unit = { + val cobolOptions = caseInsensitiveOptions.toMap + + val cobolParameters = CobolParametersParser.parse(new Parameters(cobolOptions), isWriter = true) + CobolParametersValidator.checkSanity(cobolParameters) + + val readerParameters = CobolParametersParser.getReaderProperties(cobolParameters, None) + CobolParametersValidator.validateParametersForWriting(readerParameters) + + val copybookContent = CopybookContentLoader.load(cobolParameters, spark.sparkContext.hadoopConfiguration) + val cobolSchema = CobolSchema.fromReaderParameters(copybookContent, readerParameters) + val combiner = RecordCombinerSelector.selectCombiner(cobolSchema, readerParameters) + combiner.combine(df, cobolSchema, readerParameters) + } } diff --git a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/FixedLengthEbcdicWriterSuite.scala b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/FixedLengthEbcdicWriterSuite.scala index 8163d681e..0d217811f 100644 --- a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/FixedLengthEbcdicWriterSuite.scala +++ b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/FixedLengthEbcdicWriterSuite.scala @@ -20,6 +20,7 @@ import org.apache.hadoop.fs.Path import org.apache.spark.sql.SaveMode import org.scalatest.Assertion import org.scalatest.wordspec.AnyWordSpec +import za.co.absa.cobrix.spark.cobol.Cobrix import za.co.absa.cobrix.spark.cobol.source.base.SparkTestBase import za.co.absa.cobrix.spark.cobol.source.fixtures.{BinaryFileFixture, TextComparisonFixture} import za.co.absa.cobrix.spark.cobol.utils.SparkUtils @@ -43,6 +44,10 @@ class FixedLengthEbcdicWriterSuite extends AnyWordSpec with SparkTestBase with B val path = new Path(tempDir, "writer1") + Cobrix.validateOptions + .option("copybook_contents", copybookContents) + .validateSave(df) + df.coalesce(1) .orderBy("A") .write @@ -87,6 +92,11 @@ class FixedLengthEbcdicWriterSuite extends AnyWordSpec with SparkTestBase with B val path = new Path(tempDir, "writer1") + Cobrix.validateOptions + .option("copybook_contents", copybookContents) + .option("ebcdic_code_page", "cp1144") + .validateSave(df) + df.coalesce(1) .orderBy("A") .write @@ -465,6 +475,11 @@ class FixedLengthEbcdicWriterSuite extends AnyWordSpec with SparkTestBase with B .option("copybook_contents", copybookContents) .save(path.toString) + assertThrows[IllegalArgumentException] { + Cobrix.validateOptions + .validateSave(df) + } + assertThrows[IllegalArgumentException] { df.write .format("cobol") @@ -562,6 +577,12 @@ class FixedLengthEbcdicWriterSuite extends AnyWordSpec with SparkTestBase with B val path = new Path(tempDir, "writer_redefines_conflict") + Cobrix.validateOptions + .option("copybook_contents", copybookContentsWithRedefines) + .option("write_strict_redefines", "true") + .option("pedantic", "true") + .validateSave(df) + val thrown = intercept[Throwable] { df.coalesce(1) .write