Transformer: Support old output file hierarchy (close #1314)

config/transformer/aws/transformer.batch.config.reference.hocon

@@ -166,6 +166,10 @@
     # When enabled, event's atomic fields are truncated (based on the length limits from the atomic JSON schema) before transformation.
     # Optional, default "false".
     "truncateAtomicFields": false
+
+    # Use old directory structure, i.e. vendor/name/format/model , for transformed events or not
+    # This should be enabled during upgrade from older versions of the loader
+    "legacyPartitioning": false
   }
 
   # Observability and reporting options

config/transformer/aws/transformer.kinesis.config.reference.hocon

@@ -234,6 +234,10 @@
 
     # When enabled, event's atomic fields are truncated (based on the length limits from the atomic JSON schema) before transformation.
     # Optional, default "false".
-    "truncateAtomicFields": false 
+    "truncateAtomicFields": false
+
+    # Use old directory structure, i.e. vendor/name/format/model , for transformed events or not
+    # This should be enabled during upgrade from older versions of the loader
+    "legacyPartitioning": false
   }
 }

config/transformer/azure/transformer.kafka.config.reference.hocon

@@ -150,5 +150,9 @@
     # When enabled, event's atomic fields are truncated (based on the length limits from the atomic JSON schema) before transformation.
     # Optional, default "false".
     "truncateAtomicFields": false
+
+    # Use old directory structure, i.e. vendor/name/format/model , for transformed events or not
+    # This should be enabled during upgrade from older versions of the loader
+    "legacyPartitioning": false
   }
 }

config/transformer/gcp/transformer.pubsub.config.reference.hocon

@@ -143,5 +143,9 @@
     # When enabled, event's atomic fields are truncated (based on the length limits from the atomic JSON schema) before transformation.
     # Optional, default "false".
     "truncateAtomicFields": false
+
+    # Use old directory structure, i.e. vendor/name/format/model , for transformed events or not
+    # This should be enabled during upgrade from older versions of the loader
+    "legacyPartitioning": false
   }
 }

modules/common-transformer-stream/src/main/resources/application.conf

@@ -24,7 +24,8 @@
   "featureFlags": {
     "legacyMessageFormat": false,
     "enableMaxRecordsPerFile": true,
-    "truncateAtomicFields": false
+    "truncateAtomicFields": false,
+    "legacyPartitioning": false
   }
 
   "monitoring": {

modules/common-transformer-stream/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/stream/common/Processing.scala

@@ -100,7 +100,7 @@ object Processing {
       source
         .through(transform(transformer, config.validations, processor))
         .through(incrementMetrics(resources.metrics))
-        .through(handleTransformResult(transformer))
+        .through(handleTransformResult(transformer, config.featureFlags.legacyPartitioning))
         .through(windowing)
 
     val sink: Pipe[F, Record[Window, List[(SinkPath, Transformed.Data)], State[C]], Unit] =
@@ -226,14 +226,15 @@ object Processing {
    * to where it should sink. Processes in batches for efficiency.
    */
   def handleTransformResult[F[_], C: Checkpointer[F, *]](
-    transformer: Transformer[F]
+    transformer: Transformer[F],
+    legacyPartitioning: Boolean
   ): Pipe[F, TransformationResults[C], SerializationResults[C]] =
     _.map { case (items, checkpointer) =>
       val state = State.fromEvents(items).withCheckpointer(checkpointer)
       val mapped = items.flatMap(
         _.fold(
-          bad => transformer.badTransform(bad).split :: Nil,
-          success => success.output.map(_.split)
+          bad => transformer.badTransform(bad).split(legacyPartitioning) :: Nil,
+          success => success.output.map(_.split(legacyPartitioning))
         )
       )
       (mapped, state)
@@ -255,11 +256,15 @@ object Processing {
     }
 
   implicit class TransformedOps(t: Transformed) {
-    def getPath: SinkPath = t match {
+    def getPath(legacyPartitioning: Boolean): SinkPath = t match {
       case p: Transformed.Shredded =>
-        val suffix = Some(
-          s"vendor=${p.vendor}/name=${p.name}/format=${p.format.path.toLowerCase}/model=${p.model}/revision=${p.revision}/addition=${p.addition}"
-        )
+        val suffix =
+          if (legacyPartitioning)
+            Some(s"vendor=${p.vendor}/name=${p.name}/format=${p.format.path.toLowerCase}/model=${p.model}/")
+          else
+            Some(
+              s"vendor=${p.vendor}/name=${p.name}/format=${p.format.path.toLowerCase}/model=${p.model}/revision=${p.revision}/addition=${p.addition}"
+            )
         val pathType = if (p.isGood) SinkPath.PathType.Good else SinkPath.PathType.Bad
         SinkPath(suffix, pathType)
       case p: Transformed.WideRow =>
@@ -269,6 +274,6 @@ object Processing {
       case _: Transformed.Parquet =>
         SinkPath(None, SinkPath.PathType.Good)
     }
-    def split: (SinkPath, Transformed.Data) = (getPath, t.data)
+    def split(legacyPartitioning: Boolean): (SinkPath, Transformed.Data) = (getPath(legacyPartitioning), t.data)
   }
 }

modules/common-transformer-stream/src/test/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/stream/common/processing/ShredTsvProcessingSpec.scala

@@ -24,7 +24,7 @@ class ShredTsvProcessingSpec extends BaseProcessingSpec {
             inputEventsPath = "/processing-spec/1/input/events"
           )
 
-          val config = TransformerConfig(appConfig(outputDirectory), igluConfig)
+          val config = TransformerConfig(appConfig(outputDirectory, false), igluConfig)
 
           for {
             output <- process(inputStream, config)
@@ -84,7 +84,7 @@ class ShredTsvProcessingSpec extends BaseProcessingSpec {
             inputEventsPath = "/processing-spec/3/input/events"
           )
 
-          val config = TransformerConfig(appConfig(outputDirectory), igluConfig)
+          val config = TransformerConfig(appConfig(outputDirectory, false), igluConfig)
 
           for {
             output <- process(inputStream, config)
@@ -114,11 +114,71 @@ class ShredTsvProcessingSpec extends BaseProcessingSpec {
         }
         .unsafeRunSync()
     }
+
+    "respect legacyPartitioning flag" in {
+      temporaryDirectory
+        .use { outputDirectory =>
+          val inputStream = InputEventsProvider.eventStream(
+            inputEventsPath = "/processing-spec/1/input/events"
+          )
+
+          val config = TransformerConfig(appConfig(outputDirectory, true), igluConfig)
+
+          for {
+            output <- process(inputStream, config)
+            actualAtomicRows <-
+              readStringRowsFrom(
+                Path(
+                  outputDirectory.toString +
+                    s"/run=1970-01-01-10-30-00-${AppId.appId}/output=good/vendor=com.snowplowanalytics.snowplow/name=atomic/format=tsv/model=1"
+                )
+              )
+            actualOptimizelyRows <-
+              readStringRowsFrom(
+                Path(
+                  outputDirectory.toString +
+                    s"/run=1970-01-01-10-30-00-${AppId.appId}/output=good/vendor=com.optimizely/name=state/format=tsv/model=1"
+                )
+              )
+            actualConsentRows <-
+              readStringRowsFrom(
+                Path(
+                  outputDirectory.toString +
+                    s"/run=1970-01-01-10-30-00-${AppId.appId}/output=good/vendor=com.snowplowanalytics.snowplow/name=consent_document/format=tsv/model=1"
+                )
+              )
+            actualBadRows <-
+              readStringRowsFrom(
+                Path(
+                  outputDirectory.toString +
+                    s"/run=1970-01-01-10-30-00-${AppId.appId}/output=bad/vendor=com.snowplowanalytics.snowplow.badrows/name=loader_parsing_error/format=json/model=2"
+                )
+              )
+
+            expectedCompletionMessage <- readMessageFromResource("/processing-spec/1/output/good/tsv/completion.json", outputDirectory)
+            expectedAtomicRows <- readLinesFromResource("/processing-spec/1/output/good/tsv/com.snowplowanalytics.snowplow-atomic")
+            expectedOptimizelyRows <- readLinesFromResource("/processing-spec/1/output/good/tsv/com.optimizely-state")
+            expectedConsentRows <-
+              readLinesFromResource("/processing-spec/1/output/good/tsv/com.snowplowanalytics.snowplow-consent_document")
+            expectedBadRows <- readLinesFromResource("/processing-spec/1/output/bad")
+          } yield {
+            removeAppId(output.completionMessages.toList) must beEqualTo(Vector(expectedCompletionMessage))
+            output.checkpointed must beEqualTo(1)
+
+            assertStringRows(removeAppId(actualAtomicRows), expectedAtomicRows)
+            assertStringRows(removeAppId(actualOptimizelyRows), expectedOptimizelyRows)
+            assertStringRows(removeAppId(actualConsentRows), expectedConsentRows)
+
+            assertStringRows(removeAppId(actualBadRows), expectedBadRows)
+          }
+        }
+        .unsafeRunSync()
+    }
   }
 }
 
 object ShredTsvProcessingSpec {
-  private val appConfig = (outputPath: Path) => s"""|{
+  private val appConfig = (outputPath: Path, legacyPartitioning: Boolean) => s"""|{
         | "input": {
         |   "type": "pubsub"
         |   "subscription": "projects/project-id/subscriptions/subscription-id"
@@ -137,6 +197,9 @@ object ShredTsvProcessingSpec {
         |   "region": "eu-central-1"
         | }
         | "windowing": "1 minute"
+        | "featureFlags": {
+        |    "legacyPartitioning": $legacyPartitioning
+        |  }
         | "formats": {
         |   "transformationType": "shred"
         |   "default": "TSV"

modules/common-transformer-stream/src/test/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/stream/common/sinks/TransformingSpec.scala

@@ -42,11 +42,11 @@ class TransformingSpec extends Specification {
       val testFileNameMap = List(
         Transformed.Shredded
           .Tabular("com.snowplowanalytics.snowplow", "atomic", 1, 0, 0, dummyTransformedData)
-          .getPath -> "com.snowplowanalytics.snowplow-atomic",
+          .getPath(false) -> "com.snowplowanalytics.snowplow-atomic",
         Transformed.Shredded
           .Tabular("com.snowplowanalytics.snowplow", "consent_document", 1, 0, 0, dummyTransformedData)
-          .getPath -> "com.snowplowanalytics.snowplow-consent_document",
-        Transformed.Shredded.Tabular("com.optimizely", "state", 1, 0, 0, dummyTransformedData).getPath -> "com.optimizely-state"
+          .getPath(false) -> "com.snowplowanalytics.snowplow-consent_document",
+        Transformed.Shredded.Tabular("com.optimizely", "state", 1, 0, 0, dummyTransformedData).getPath(false) -> "com.optimizely-state"
       ).toMap
 
       val expectedTransformedMap =
@@ -149,7 +149,7 @@ object TransformingSpec {
 
     val eventStream = parsedEventStream(resourcePath)
       .through(Processing.transform(transformer, validations, TestProcessor))
-      .through(Processing.handleTransformResult(transformer))
+      .through(Processing.handleTransformResult(transformer, legacyPartitioning = false))
 
     val transformed = eventStream.compile.toList.unsafeRunSync().flatMap(_._1)
     (transformed.flatMap(_.getGood), transformed.flatMap(_.getBad))

modules/common/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/common/config/TransformerConfig.scala

@@ -144,7 +144,8 @@ object TransformerConfig {
     legacyMessageFormat: Boolean,
     sparkCacheEnabled: Option[Boolean],
     enableMaxRecordsPerFile: Boolean,
-    truncateAtomicFields: Boolean
+    truncateAtomicFields: Boolean,
+    legacyPartitioning: Boolean
   )
 
   object FeatureFlags {

modules/transformer-batch/src/main/resources/application.conf

@@ -41,6 +41,7 @@
   "featureFlags": {
     "legacyMessageFormat": false,
     "enableMaxRecordsPerFile": false,
-    "truncateAtomicFields": false
+    "truncateAtomicFields": false,
+    "legacyPartitioning": false
   }
 }

modules/transformer-batch/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/batch/ShredJob.scala

@@ -294,7 +294,8 @@ object ShredJob {
     unshredded.foreach { folder =>
       System.out.println(s"Batch Transformer: processing $folder")
       val transformer = config.formats match {
-        case f: TransformerConfig.Formats.Shred => Transformer.ShredTransformer(resolverConfig, f, maxRecordsPerFile = 0)
+        case f: TransformerConfig.Formats.Shred =>
+          Transformer.ShredTransformer(resolverConfig, f, maxRecordsPerFile = 0, config.featureFlags.legacyPartitioning)
         case TransformerConfig.Formats.WideRow.JSON => Transformer.WideRowJsonTransformer()
         case TransformerConfig.Formats.WideRow.PARQUET =>
           val resolver = IgluSingleton.get(resolverConfig)

modules/transformer-batch/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/batch/Transformer.scala

@@ -65,7 +65,8 @@ object Transformer {
   case class ShredTransformer(
     resolverConfig: ResolverConfig,
     formats: Formats.Shred,
-    maxRecordsPerFile: Long
+    maxRecordsPerFile: Long,
+    legacyPartitioning: Boolean
   ) extends Transformer[TypesInfo.Shredded.Type] {
     val typesAccumulator = new TypesAccumulator[TypesInfo.Shredded.Type]
     val timestampsAccumulator: TimestampsAccumulator = new TimestampsAccumulator
@@ -119,7 +120,10 @@ object Transformer {
       outFolder: Folder,
       maxRecordsPerFile: Long
     ): Unit =
-      Sink.writeShredded(spark, compression, transformed.flatMap(_.shredded), outFolder)
+      if (legacyPartitioning)
+        Sink.legacyWriteShredded(spark, compression, transformed.flatMap(_.legacyShredded), outFolder)
+      else
+        Sink.writeShredded(spark, compression, transformed.flatMap(_.shredded), outFolder)
 
     def register(sc: SparkContext): Unit = {
       sc.register(typesAccumulator)
@@ -215,6 +219,7 @@ object Transformer {
 
   type WideRowTuple = (String, String)
   type ShreddedTuple = (String, String, String, String, Int, Int, Int, String)
+  type LegacyShreddedTuple = (String, String, String, String, Int, String)
 
   private implicit class TransformedOps(t: Transformed) {
     def wideRow: Option[WideRowTuple] = t match {
@@ -231,6 +236,13 @@ object Transformer {
       case _ => None
     }
 
+    def legacyShredded: Option[LegacyShreddedTuple] = t match {
+      case p: Transformed.Shredded =>
+        val outputType = if (p.isGood) "good" else "bad"
+        (outputType, p.vendor, p.name, p.format.path, p.model, p.data.value).some
+      case _ => None
+    }
+
     def parquet: Option[List[Any]] = t match {
       case p: Transformed.Parquet => p.data.value.map(_.value).map(extractFieldValue).some
       case _ => None

modules/transformer-batch/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/batch/spark/Sink.scala

@@ -11,13 +11,30 @@ import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{DataFrameWriter, Row, SaveMode, SparkSession}
 import org.apache.spark.sql.types.StructType
 import com.snowplowanalytics.snowplow.rdbloader.common.config.TransformerConfig.Compression
+import com.snowplowanalytics.snowplow.rdbloader.transformer.batch.Transformer.{LegacyShreddedTuple, ShreddedTuple}
 
 object Sink {
 
+  def legacyWriteShredded(
+    spark: SparkSession,
+    compression: Compression,
+    data: RDD[LegacyShreddedTuple],
+    outFolder: String
+  ): Unit = {
+    import spark.implicits._
+    data
+      .toDF("output", "vendor", "name", "format", "model", "data")
+      .write
+      .withCompression(compression)
+      .partitionBy("output", "vendor", "name", "format", "model")
+      .mode(SaveMode.Append)
+      .text(outFolder)
+  }
+
   def writeShredded(
     spark: SparkSession,
     compression: Compression,
-    data: RDD[(String, String, String, String, Int, Int, Int, String)],
+    data: RDD[ShreddedTuple],
     outFolder: String
   ): Unit = {
     import spark.implicits._

modules/transformer-batch/src/test/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/batch/ConfigSpec.scala

@@ -155,7 +155,7 @@ object TransformerConfigSpec {
     Some(Duration.create("14 days").asInstanceOf[FiniteDuration]),
     Some(Config.RunInterval.IntervalInstant(Instant.parse("2021-12-10T18:34:52.00Z")))
   )
-  val exampleDefaultFeatureFlags = TransformerConfig.FeatureFlags(false, None, false, false)
+  val exampleDefaultFeatureFlags = TransformerConfig.FeatureFlags(false, None, false, false, false)
   val exampleValidations = Validations(Some(Instant.parse("2021-11-18T11:00:00.00Z")))
   val emptyValidations = Validations(None)
   val exampleSkipSchemas = List(