#788 Unify processor builders between the parser and spark module.

yruslan · yruslan · commit 12b38457e384 · 2025-10-10T09:34:51.000+02:00
diff --git a/README.md b/README.md
@@ -1668,15 +1668,17 @@ The EBCDIC processor allows processing files by replacing value of fields withou
 
 The processing does not require Spark. A processing application can have only the COBOL parser as a dependency (`cobol-parser`).
 
-Here is an example usage:
+Here is an example usage (using streams of bytes):
 ```scala
 val is = new FSStream(inputFile)
 val os = new FileOutputStream(outputFile)
-val copybookContents = "...some copybook..."
 val builder = CobolProcessor.builder(copybookContents)
 
+val builder = CobolProcessor.builder
+  .withCopybookContents("...some copybook...")
+
 val processor = new RawRecordProcessor {
-  override def processRecord(copybook: Copybook, options: Map[String, String], record: Array[Byte], offset: Long): Array[Byte] = {
+  override def processRecord(record: Array[Byte], ctx: CobolProcessorContext): Array[Byte] = {
     // The transformation logic goes here
     val value = copybook.getFieldValueByName("some_field", record, 0)
     // Change the field v
@@ -1688,10 +1690,53 @@ val processor = new RawRecordProcessor {
   }
 }
 
-builder.build().process(is, os)(processor)
+val count = builder.build().process(is, os)(processor)
+```
+
+Here is an example usage (using paths):
+```scala
+val count = CobolProcessor.builder
+  .withCopybookContents(copybook)
+  .withRecordProcessor { (record: Array[Byte], ctx: CobolProcessorContext) =>
+    // The transformation logic goes here
+    val value = copybook.getFieldValueByName("some_field", record, 0)
+    // Change the field v
+    // val newValue = ...
+    // Write the changed value back
+    copybook.setFieldValueByName("some_field", record, newValue, 0)
+    // Return the changed record     
+    record
+  }
+  .load(inputFile)
+  .save(outputFile)
 ```
 
 
+## EBCDIC Spark Processor (experimental)
+This allows in-place processing of data retaining original format in parallel uring RDDs under the hood.
+
+Here is an example usage:
+```scala
+import za.co.absa.cobrix.spark.cobol.SparkCobolProcessor
+
+val copybookContents = "...some copybook..."
+
+SparkCobolProcessor.builder
+  .withCopybookContents(copybook)
+  .withRecordProcessor { (record: Array[Byte], ctx: CobolProcessorContext) =>
+    // The transformation logic goes here
+    val value = ctx.copybook.getFieldValueByName("some_field", record, 0)
+    // Change the field v
+    // val newValue = ...
+    // Write the changed value back
+    ctx.copybook.setFieldValueByName("some_field", record, newValue, 0)
+    // Return the changed record     
+    record
+  }
+  .load(inputPath)
+  .save(outputPath)
+```
+
 ## EBCDIC Writer (experimental)
 
 Cobrix's EBCDIC writer is an experimental feature that allows writing Spark DataFrames as EBCDIC mainframe files.
diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/processor/CobolProcessor.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/processor/CobolProcessor.scala
@@ -16,12 +16,13 @@
 
 package za.co.absa.cobrix.cobol.processor
 
+import za.co.absa.cobrix.cobol.parser.Copybook
 import za.co.absa.cobrix.cobol.processor.impl.CobolProcessorImpl
 import za.co.absa.cobrix.cobol.reader.parameters.{CobolParametersParser, Parameters, ReaderParameters}
 import za.co.absa.cobrix.cobol.reader.schema.CobolSchema
-import za.co.absa.cobrix.cobol.reader.stream.SimpleStream
+import za.co.absa.cobrix.cobol.reader.stream.{FSStream, SimpleStream}
 
-import java.io.OutputStream
+import java.io.{BufferedInputStream, BufferedOutputStream, FileOutputStream, OutputStream}
 import scala.collection.mutable
 
 
@@ -45,14 +46,58 @@ trait CobolProcessor {
 }
 
 object CobolProcessor {
-  class CobolProcessorBuilder(copybookContents: String) {
+  class CobolProcessorBuilder {
     private val caseInsensitiveOptions = new mutable.HashMap[String, String]()
+    private var copybookContentsOpt: Option[String] = None
+    private var rawRecordProcessorOpt: Option[RawRecordProcessor] = None
 
     def build(): CobolProcessor = {
+      if (copybookContentsOpt.isEmpty) {
+        throw new IllegalArgumentException("Copybook contents must be provided.")
+      }
+
+      val readerParameters = getReaderParameters
+      val cobolSchema = getCobolSchema(readerParameters)
+
+      new CobolProcessorImpl(readerParameters, cobolSchema.copybook, copybookContentsOpt.get, caseInsensitiveOptions.toMap)
+    }
+
+    def load(path: String): CobolProcessorLoader = {
+      val file = new java.io.File(path)
+      if (!file.exists) {
+        throw new IllegalArgumentException(s"Path $path does not exist.")
+      }
+
+      if (file.isDirectory) {
+        throw new IllegalArgumentException(s"Path $path should be a file, not a directory.")
+      }
+
+      if (copybookContentsOpt.isEmpty) {
+        throw new IllegalArgumentException("Copybook contents must be provided.")
+      }
+
+      if (rawRecordProcessorOpt.isEmpty) {
+        throw new IllegalArgumentException("A RawRecordProcessor must be provided.")
+      }
+
+      if (rawRecordProcessorOpt.isEmpty) {
+        throw new IllegalArgumentException("A RawRecordProcessor must be provided.")
+      }
+
       val readerParameters = getReaderParameters
       val cobolSchema = getCobolSchema(readerParameters)
 
-      new CobolProcessorImpl(readerParameters, cobolSchema.copybook, copybookContents, caseInsensitiveOptions.toMap)
+      new CobolProcessorLoader(path, copybookContentsOpt.get, cobolSchema.copybook, rawRecordProcessorOpt.get, readerParameters, caseInsensitiveOptions.toMap)
+    }
+
+    def withCopybookContents(copybookContents: String): CobolProcessorBuilder = {
+      copybookContentsOpt = Option(copybookContents)
+      this
+    }
+
+    def withRecordProcessor(processor: RawRecordProcessor): CobolProcessorBuilder = {
+      rawRecordProcessorOpt = Option(processor)
+      this
     }
 
     /**
@@ -80,7 +125,7 @@ object CobolProcessor {
     }
 
     private[processor] def getCobolSchema(readerParameters: ReaderParameters): CobolSchema = {
-      CobolSchema.fromReaderParameters(Seq(copybookContents), readerParameters)
+      CobolSchema.fromReaderParameters(Seq(copybookContentsOpt.get), readerParameters)
     }
 
     private[processor] def getReaderParameters: ReaderParameters = {
@@ -92,7 +137,57 @@ object CobolProcessor {
     private[processor] def getOptions: Map[String, String] = caseInsensitiveOptions.toMap
   }
 
-  def builder(copybookContent: String): CobolProcessorBuilder = {
-    new CobolProcessorBuilder(copybookContent)
+  class CobolProcessorLoader(fileToProcess: String,
+                             copybookContents: String,
+                             copybook: Copybook,
+                             rawRecordProcessor: RawRecordProcessor,
+                             readerParameters: ReaderParameters,
+                             options: Map[String, String]) {
+    def save(outputFile: String): Long = {
+      val processor = new CobolProcessorImpl(readerParameters, copybook, copybookContents, options)
+
+      val ifs = new FSStream(fileToProcess)
+      val ofs = new BufferedOutputStream(new FileOutputStream(outputFile))
+
+      var originalException: Throwable = null
+
+      val recordCount = try {
+        processor.process(ifs, ofs)(rawRecordProcessor)
+      } catch {
+        case ex: Throwable =>
+          originalException = ex
+          0L
+      } finally {
+        try {
+          ifs.close()
+        } catch {
+          case e: Throwable =>
+            if (originalException != null) {
+              originalException.addSuppressed(e)
+            } else {
+              originalException = e
+            }
+        }
+
+        try {
+          ofs.close()
+        } catch {
+          case e: Throwable =>
+            if (originalException != null) {
+              originalException.addSuppressed(e)
+            } else {
+              originalException = e
+            }
+        }
+      }
+
+      if (originalException != null) throw originalException
+
+      recordCount
+    }
+  }
+
+  def builder: CobolProcessorBuilder = {
+    new CobolProcessorBuilder
   }
 }
diff --git a/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/base/BinaryFileFixture.scala b/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/base/BinaryFileFixture.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.cobol.base
+
+import java.nio.file.{Files, Path, Paths}
+import java.util.Comparator
+import java.util.function.Consumer
+
+/**
+  * This fixture adds ability for a unit test to create temporary files for using them in the tests.
+  */
+trait BinaryFileFixture {
+  def withTempDirectory(prefix: String)(f: String => Unit): Unit = {
+    val tmpPath = Files.createTempDirectory(prefix)
+    val pathStr = tmpPath.toAbsolutePath.toString
+
+    f(pathStr)
+
+    Files.walk(tmpPath)
+      .sorted(Comparator.reverseOrder())
+      .forEach(new Consumer[Path] {
+        override def accept(f: Path): Unit = Files.delete(f)
+      })
+  }
+
+  def writeBinaryFile(filePath: String, content: Array[Byte]): Unit = {
+    Files.write(Paths.get(filePath), content)
+  }
+
+  def readBinaryFile(filePath: String): Array[Byte] = {
+    Files.readAllBytes(Paths.get(filePath))
+  }
+
+  private def hex2bytes(hex: String): Array[Byte] = {
+    val compactStr = hex.replaceAll("\\s", "")
+    compactStr.sliding(2, 2).toArray.map(Integer.parseInt(_, 16).toByte)
+  }
+}
diff --git a/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/processor/CobolProcessorBuilderSuite.scala b/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/processor/CobolProcessorBuilderSuite.scala
@@ -17,15 +17,15 @@
 package za.co.absa.cobrix.cobol.processor
 
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.cobrix.cobol.base.BinaryFileFixture
 import za.co.absa.cobrix.cobol.mock.ByteStreamMock
-import za.co.absa.cobrix.cobol.parser.Copybook
 import za.co.absa.cobrix.cobol.parser.recordformats.RecordFormat
-import za.co.absa.cobrix.cobol.reader.extractors.raw.{FixedRecordLengthRawRecordExtractor, TextFullRecordExtractor}
 import za.co.absa.cobrix.cobol.reader.parameters.ReaderParameters
 
 import java.io.ByteArrayOutputStream
+import java.nio.file.Paths
 
-class CobolProcessorBuilderSuite extends AnyWordSpec {
+class CobolProcessorBuilderSuite extends AnyWordSpec with BinaryFileFixture {
   private val copybook =
     """      01 RECORD.
       |         05  T     PIC X.
@@ -35,7 +35,8 @@ class CobolProcessorBuilderSuite extends AnyWordSpec {
     "process an input data stream into an output stream" in {
       val is = new ByteStreamMock(Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
       val os = new ByteArrayOutputStream(10)
-      val builder = CobolProcessor.builder(copybook)
+      val builder = CobolProcessor.builder
+        .withCopybookContents(copybook)
 
       val processor = new RawRecordProcessor {
         override def processRecord(record: Array[Byte], ctx: CobolProcessorContext): Array[Byte] = {
@@ -55,9 +56,39 @@ class CobolProcessorBuilderSuite extends AnyWordSpec {
     }
   }
 
+  "load and save" should {
+    "process files as expected" in {
+      withTempDirectory("cobol_processor") { tempDir =>
+        val inputFile = Paths.get(tempDir, "input.dat").toString
+        val outputFile = Paths.get(tempDir, "output.dat").toString
+
+        writeBinaryFile(inputFile, Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
+
+        val count = CobolProcessor.builder
+          .withCopybookContents(copybook)
+          .withRecordProcessor(new RawRecordProcessor {
+            override def processRecord(record: Array[Byte], ctx: CobolProcessorContext): Array[Byte] = {
+              record.map(v => (v - 1).toByte)
+            }
+          })
+          .load(inputFile)
+          .save(outputFile)
+
+        val outputArray = readBinaryFile(outputFile)
+
+        assert(count == 4)
+        assert(outputArray.head == -16)
+        assert(outputArray(1) == -15)
+        assert(outputArray(2) == -14)
+        assert(outputArray(3) == -13)
+      }
+    }
+  }
+
   "getCobolSchema" should {
     "return the schema of the copybook provided" in {
-      val builder = CobolProcessor.builder(copybook)
+      val builder = CobolProcessor.builder
+        .withCopybookContents(copybook)
 
       val cobolSchema = builder.getCobolSchema(ReaderParameters())
 
@@ -67,7 +98,8 @@ class CobolProcessorBuilderSuite extends AnyWordSpec {
 
   "getReaderParameters" should {
     "return a reader according to passed options" in {
-      val builder = CobolProcessor.builder(copybook)
+      val builder = CobolProcessor.builder
+        .withCopybookContents(copybook)
         .option("record_format", "D")
 
       assert(builder.getReaderParameters.recordFormat == RecordFormat.AsciiText)
diff --git a/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/processor/impl/CobolProcessorImplSuite.scala b/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/processor/impl/CobolProcessorImplSuite.scala
@@ -32,7 +32,9 @@ class CobolProcessorImplSuite extends AnyWordSpec {
   "getRecordExtractor" should {
     "work for an fixed-record-length files" in {
       val stream = new ByteStreamMock(Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
-      val processor = CobolProcessor.builder(copybook).build().asInstanceOf[CobolProcessorImpl]
+      val processor = CobolProcessor.builder
+        .withCopybookContents(copybook)
+        .build().asInstanceOf[CobolProcessorImpl]
 
       val ext = processor.getRecordExtractor(ReaderParameters(recordLength = Some(2), options = Map("test" -> "option")), stream)
 
@@ -46,7 +48,9 @@ class CobolProcessorImplSuite extends AnyWordSpec {
 
     "work for an variable-record-length files" in {
       val stream = new ByteStreamMock(Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
-      val processor = CobolProcessor.builder(copybook).build().asInstanceOf[CobolProcessorImpl]
+      val processor = CobolProcessor.builder
+        .withCopybookContents(copybook)
+        .build().asInstanceOf[CobolProcessorImpl]
 
       val ext = processor.getRecordExtractor(ReaderParameters(
         recordFormat = RecordFormat.VariableLength,
@@ -58,7 +62,9 @@ class CobolProcessorImplSuite extends AnyWordSpec {
 
     "throw an exception on a non-supported record format for processing" in {
       val stream = new ByteStreamMock(Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
-      val processor = CobolProcessor.builder(copybook).build().asInstanceOf[CobolProcessorImpl]
+      val processor = CobolProcessor.builder
+        .withCopybookContents(copybook)
+        .build().asInstanceOf[CobolProcessorImpl]
 
       val ex = intercept[IllegalArgumentException] {
         processor.getRecordExtractor(ReaderParameters(
diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessor.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessor.scala
diff --git a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessorSuite.scala b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessorSuite.scala
