#875 Add the ability to validate options to spark-cobol without reading data files

README.md

@@ -829,6 +829,26 @@ 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 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
 
 The following code pages are supported:

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()
+  }
 }

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)
   }
 }

spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/builder/ValidatorBuilder.scala

@@ -0,0 +1,101 @@
+/*
+ * 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.{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
+
+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
+  }
+
+
+  /**
+    * 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 validateLoad(filePaths: String*): Unit = {
+    val sqlContext = spark.sqlContext
+
+    // Validating options compatibility
+    val cobolOptions = caseInsensitiveOptions.toMap + (PARAM_SOURCE_PATHS -> filePaths.mkString(","))
+    CobolParametersValidator.validateOrThrow(cobolOptions, spark.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
+    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)
+
+    // 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)
+  }
+}

spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/CobolRelation.scala

@@ -104,6 +104,15 @@ 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
+    }
+  }
 }
 
 object CobolRelation {

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)
+  }
 }

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)
     }
   }
 

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()))

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],

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())
   }
 

spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/HDFSUtils.scala → 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
 

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.")