Extending HDFS Lineage Dispatcher to support writing lineage files to a centralised location

core/src/main/resources/spline.default.yaml

@@ -146,7 +146,27 @@ spline:
       fileBufferSize: 4096
       # file/directory permissions, provided as umask string, either in octal or symbolic format
       filePermissions: 777
-
+      customLineagePath:
+      # OPTIONAL: Custom path for centralized lineage storage.
+      # If left empty, null, or not specified → DEFAULT MODE: lineage written alongside target data files
+      # If set to a path → CENTRALIZED MODE: all lineage written to this location with unique filenames
+      #
+      # CENTRALIZED MODE filename format: {timestamp}_{fileName}_{appId}
+      # - timestamp: Human-readable UTC timestamp (yyyy-MM-dd_HH-mm-ss-SSS) for natural chronological sorting and easy filtering
+      #   Example: 2025-10-12_14-30-45-123
+      # - fileName: The configured fileName value (e.g., "my_file.parq_LINEAGE")
+      # - appId: Spark application ID for traceability to specific runs
+      #   Example: app-20251012143045-0001
+      #
+      # More examples:
+      # - Local: customLineagePath: /my/centralized/lineage
+      #   Output: /my/centralized/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
+      # - S3: customLineagePath: s3://my-bucket/lineage
+      #   Output: s3://my-bucket/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
+      # - GCS: customLineagePath: gs://my-bucket/lineage
+      #   Output: gs://my-bucket/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
+      # - HDFS: customLineagePath: hdfs://cluster/lineage
+      #   Output: hdfs://cluster/lineage/2025-10-12_14-30-45-123_my_file.parq_LINEAGE_app-20251012143045-0001
     # -------------------------------------------
     # Open Lineage HTTP dispatcher
     # -------------------------------------------

core/src/main/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcher.scala

@@ -30,6 +30,8 @@ import za.co.absa.spline.harvester.json.HarvesterJsonSerDe
 import za.co.absa.spline.producer.model.{ExecutionEvent, ExecutionPlan}
 
 import java.net.URI
+import java.time.format.DateTimeFormatter
+import java.time.{Instant, ZoneId}
 import scala.concurrent.blocking
 
 /**
@@ -40,16 +42,34 @@ import scala.concurrent.blocking
  * for every generic use case in a real production application.
  *
  * It is NOT thread-safe, strictly synchronous assuming a predefined order of method calls: `send(plan)` and then `send(event)`
+ *
+ * TWO MODES OF OPERATION:
+ *
+ * 1. DEFAULT MODE (customLineagePath = None, null, or empty string):
+ *    Lineage files are written alongside the target data files.
+ *    Example: Writing to /data/output/file.parquet creates /data/output/_LINEAGE
+ *
+ * 2. CENTRALIZED MODE (customLineagePath set to a valid path):
+ *    All lineage files are written to a single centralized location with unique filenames.
+ *    Filename format: {timestamp}_{fileName}_{appId}
+ *    - timestamp: Human-readable UTC timestamp (yyyy-MM-dd_HH-mm-ss-SSS) for chronological sorting and filtering
+ *    - fileName: The configured fileName value (e.g., "my_file.parq_LINEAGE")
+ *    - appId: Spark application ID for traceability
+ *
+ *    The timestamp-first format ensures natural chronological sorting and easy date-based filtering.
+ *    Parent directories are automatically created with proper permissions for multi-user access (HDFS/local).
+ *    For object storage (S3, GCS, Azure), directory creation is skipped since they use key prefixes.
  */
 @Experimental
-class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSize: Int)
+class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSize: Int, customLineagePath: Option[String])
   extends LineageDispatcher
     with Logging {
 
   def this(conf: Configuration) = this(
     filename = conf.getRequiredString(FileNameKey),
     permission = new FsPermission(conf.getRequiredObject(FilePermissionsKey).toString),
-    bufferSize = conf.getRequiredInt(BufferSizeKey)
+    bufferSize = conf.getRequiredInt(BufferSizeKey),
+    customLineagePath = conf.getOptionalString(CustomLineagePathKey)
   )
 
   @volatile
@@ -67,7 +87,7 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
       throw new IllegalStateException("send(event) must be called strictly after send(plan) method with matching plan ID")
 
     try {
-      val path = s"${this._lastSeenPlan.operations.write.outputSource.stripSuffix("/")}/$filename"
+      val path = resolveLineagePath(event.planId)
       val planWithEvent = Map(
         "executionPlan" -> this._lastSeenPlan,
         "executionEvent" -> event
@@ -80,10 +100,102 @@ class HDFSLineageDispatcher(filename: String, permission: FsPermission, bufferSi
     }
   }
 
+  /**
+   * Resolves the lineage file path based on configuration.
+   * If customLineagePath is specified, lineage files are written to that centralized location.
+   * Otherwise, lineage files are written alongside the target data file (current behavior).
+   *
+   * @param planId The execution plan ID used for generating unique filenames in centralized mode
+   * @return The full path where the lineage file should be written
+   */
+  private def resolveLineagePath(planId: String): String = {
+    customLineagePath match {
+      case Some(customPath) =>
+        // Centralized mode: write to custom path with unique filename
+        val cleanCustomPath = customPath.stripSuffix("/")
+        val uniqueFilename = generateUniqueFilename(planId)
+        s"$cleanCustomPath/$uniqueFilename"
+      case None =>
+        // Default mode: write alongside target data file
+        s"${this._lastSeenPlan.operations.write.outputSource.stripSuffix("/")}/$filename"
+    }
+  }
+
+  /**
+   * Generates a unique filename for centralized lineage storage.
+   * 
+   * Format: {timestamp}_{fileName}_{appId}
+   * Example: 2025-10-12_14-30-45-123_lineage_app-20251012143045-0001
+   *
+   * This format optimizes for operational debugging use cases:
+   * - Timestamp FIRST: Ensures natural chronological sorting (most recent files appear together)
+   * - Application ID: Full traceability to specific Spark application run
+   *
+   * @param planId The execution plan ID (unused, kept for interface compatibility)
+   * @return A unique filename optimized for filtering and sorting
+   */
+  private def generateUniqueFilename(planId: String): String = {
+    val sparkContext = SparkContext.getOrCreate()
+    val appId = sparkContext.applicationId
+    val dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss-SSS").withZone(ZoneId.of("UTC"))
+    val timestamp = dateFormatter.format(Instant.now())
+    s"${timestamp}_${filename}_${appId}"
+  }
+
+  /**
+   * Ensures that parent directories exist with proper permissions for multi-user access.
+   * This is critical for centralized lineage storage where different service accounts need write access.
+   *
+   * Note: Object storage systems (S3, GCS, Azure Blob) don't have true directories - they use key prefixes.
+   * Directory creation is automatically skipped for these systems to avoid unnecessary operations.
+   *
+   * @param fs The Hadoop FileSystem
+   * @param path The target file path whose parent directories should be created
+   */
+  private def ensureParentDirectoriesExist(fs: FileSystem, path: Path): Unit = {
+    // Object storage systems (S3, GCS, Azure Blob) don't have real directories - they're just key prefixes
+    // Skip directory creation for these systems to avoid unnecessary operations
+    val fsScheme = fs.getUri.getScheme
+    val isObjectStorage = fsScheme != null && (
+      fsScheme.startsWith("s3") ||      // S3: s3, s3a, s3n
+      fsScheme.startsWith("gs") ||      // Google Cloud Storage: gs
+      fsScheme.startsWith("wasb") ||    // Azure Blob Storage: wasb, wasbs
+      fsScheme.startsWith("abfs") ||    // Azure Data Lake Storage Gen2: abfs, abfss
+      fsScheme.startsWith("adl")        // Azure Data Lake Storage Gen1: adl
+    )
+
+    if (isObjectStorage) {
+      logDebug(s"Skipping directory creation for object storage filesystem ($fsScheme) - directories are implicit key prefixes")
+      return
+    }
+
+    val parentDir = path.getParent
+    if (parentDir != null && !fs.exists(parentDir)) {
+      try {
+        // Create directories with multi-user friendly permissions to allow all service accounts to write
+        // This uses the same permission object that's already configured for files
+        val created = fs.mkdirs(parentDir, permission)
+        if (created) {
+          logInfo(s"Created parent directories: $parentDir with permissions $permission")
+        }
+      } catch {
+        case e: org.apache.hadoop.fs.FileAlreadyExistsException =>
+          // Race condition: another process created the directory - this is fine
+          logDebug(s"Directory $parentDir already exists (created by another process)")
+        case e: Exception =>
+          logWarning(s"Failed to create parent directories: $parentDir", e)
+          throw e
+      }
+    }
+  }
+
   private def persistToHadoopFs(content: String, fullLineagePath: String): Unit = blocking {
     val (fs, path) = pathStringToFsWithPath(fullLineagePath)
     logDebug(s"Opening HadoopFs output stream to $path")
 
+    // Ensure parent directories exist with proper permissions for multi-user access
+    ensureParentDirectoriesExist(fs, path)
+
     val replication = fs.getDefaultReplication(path)
     val blockSize = fs.getDefaultBlockSize(path)
     val outputStream = fs.create(path, permission, true, bufferSize, replication, blockSize, null)
@@ -104,6 +216,7 @@ object HDFSLineageDispatcher {
   private val FileNameKey = "fileName"
   private val FilePermissionsKey = "filePermissions"
   private val BufferSizeKey = "fileBufferSize"
+  private val CustomLineagePathKey = "customLineagePath"
 
   /**
    * Converts string full path to Hadoop FS and Path, e.g.

integration-tests/src/test/scala/za/co/absa/spline/harvester/dispatcher/HDFSLineageDispatcherSpec.scala

@@ -37,7 +37,7 @@ class HDFSLineageDispatcherSpec
 
   behavior of "HDFSLineageDispatcher"
 
-  it should "save lineage file to a filesystem" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+  it should "save lineage file to a filesystem in DEFAULT mode (no customLineagePath)" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
     withIsolatedSparkSession(_
       .config("spark.spline.lineageDispatcher", "hdfs")
       .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
@@ -63,4 +63,102 @@ class HDFSLineageDispatcherSpec
     }
   }
 
+  it should "save lineage file in CENTRALIZED mode with customLineagePath" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+    val centralizedPath = TempDirectory("spline_centralized_", "", pathOnly = true).deleteOnExit()
+
+    withIsolatedSparkSession(_
+      .config("spark.spline.lineageDispatcher", "hdfs")
+      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
+      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", centralizedPath.asString)
+    ) { implicit spark =>
+      withLineageTracking { captor =>
+        import spark.implicits._
+        val dummyDF = Seq((1, 2)).toDF
+        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
+
+        for {
+          (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
+        } yield {
+          // Lineage should NOT be in the destination directory
+          val lineageFileInDest = new File(destPath.asString, "_LINEAGE")
+          lineageFileInDest.exists should be(false)
+
+          // Lineage should be in centralized directory with timestamp-based filename
+          val centralizedDir = new File(centralizedPath.asString)
+          val lineageFiles = centralizedDir.listFiles()
+          lineageFiles should not be null
+          lineageFiles.length should be(1)
+
+          val lineageFile = lineageFiles(0)
+          lineageFile.length should be > 0L
+
+          // Verify filename format: {timestamp}_{fileName}_{appId}
+          val filename = lineageFile.getName
+          // Should match pattern: yyyy-MM-dd_HH-mm-ss-SSS__LINEAGE_app-...
+          filename should include("_LINEAGE_")
+          filename should startWith regex """\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}-\d{3}"""
+
+          val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
+          lineageJson should contain key "executionPlan"
+          lineageJson should contain key "executionEvent"
+          lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
+        }
+      }
+    }
+  }
+
+  it should "use DEFAULT mode when customLineagePath is empty string" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+    withIsolatedSparkSession(_
+      .config("spark.spline.lineageDispatcher", "hdfs")
+      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
+      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", "") // Empty string
+    ) { implicit spark =>
+      withLineageTracking { captor =>
+        import spark.implicits._
+        val dummyDF = Seq((1, 2)).toDF
+        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
+
+        for {
+          (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
+        } yield {
+          val lineageFile = new File(destPath.asString, "_LINEAGE")
+          lineageFile.exists should be(true)
+          lineageFile.length should be > 0L
+
+          val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
+          lineageJson should contain key "executionPlan"
+          lineageJson should contain key "executionEvent"
+          lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
+        }
+      }
+    }
+  }
+
+  it should "use DEFAULT mode when customLineagePath is whitespace only" taggedAs ignoreIf(ver"$SPARK_VERSION" < ver"2.3") in {
+    withIsolatedSparkSession(_
+      .config("spark.spline.lineageDispatcher", "hdfs")
+      .config("spark.spline.lineageDispatcher.hdfs.className", classOf[HDFSLineageDispatcher].getName)
+      .config("spark.spline.lineageDispatcher.hdfs.customLineagePath", "   ") // Whitespace only
+    ) { implicit spark =>
+      withLineageTracking { captor =>
+        import spark.implicits._
+        val dummyDF = Seq((1, 2)).toDF
+        val destPath = TempDirectory("spline_", ".parquet", pathOnly = true).deleteOnExit()
+
+        for {
+          (_, _) <- captor.lineageOf(dummyDF.write.save(destPath.asString))
+        } yield {
+          val lineageFile = new File(destPath.asString, "_LINEAGE")
+          lineageFile.exists should be(true)
+          lineageFile.length should be > 0L
+
+          val lineageJson = readFileToString(lineageFile, "UTF-8").fromJson[Map[String, Map[String, _]]]
+          lineageJson should contain key "executionPlan"
+          lineageJson should contain key "executionEvent"
+          lineageJson("executionPlan")("id") should equal(lineageJson("executionEvent")("planId"))
+        }
+      }
+    }
+  }
+
 }