GCP variant of Snowflake Loader and Stream Transformer (close #1091)

Author: spenes

.github/workflows/ci.yml

@@ -67,6 +67,7 @@ jobs:
           - snowflakeLoader
           - databricksLoader
           - transformerKinesis
+          - transformerPubsub
     steps:
       - name: Checkout Github
         uses: actions/checkout@v2
@@ -171,26 +172,17 @@ jobs:
         curl https://databricks-bi-artifacts.s3.us-east-2.amazonaws.com/simbaspark-drivers/jdbc/2.6.25/DatabricksJDBC42-2.6.25.1044.zip --output DatabricksJDBC42.jar.zip
         unzip DatabricksJDBC42.jar.zip
         cp DatabricksJDBC42-2.6.25.1044/DatabricksJDBC42.jar DatabricksJDBC42.jar
-    - name: Build redshift-loader jar
+    - name: Build jar files
       env:
         SKIP_TEST: true
-      run: sbt 'project redshiftLoader' assembly
-    - name: Build snowflake-loader jar
-      env:
-        SKIP_TEST: true
-      run: sbt 'project snowflakeLoader' assembly
-    - name: Build databricks-loader jar
-      env:
-        SKIP_TEST: true
-      run: sbt 'project databricksLoader' assembly
-    - name: Build transformer-batch jar
-      env:
-        SKIP_TEST: true
-      run: sbt 'project transformerBatch' assembly
-    - name: Build transformer-kinesis jar
-      env:
-        SKIP_TEST: true
-      run: sbt 'project transformerKinesis' assembly
+      run: |
+        sbt \
+        'project redshiftLoader; assembly' \
+        'project snowflakeLoader; assembly' \
+        'project databricksLoader; assembly' \
+        'project transformerBatch; assembly' \
+        'project transformerKinesis; assembly' \
+        'project transformerPubsub; assembly'
     - name: Get current version
       id: ver
       run: echo "::set-output name=project_version::${GITHUB_REF#refs/tags/}"
@@ -207,5 +199,6 @@ jobs:
           modules/databricks-loader/target/scala-2.12/snowplow-databricks-loader-${{ steps.ver.outputs.project_version }}.jar
           modules/transformer-batch/target/scala-2.12/snowplow-transformer-batch-${{ steps.ver.outputs.project_version }}.jar
           modules/transformer-kinesis/target/scala-2.12/snowplow-transformer-kinesis-${{ steps.ver.outputs.project_version }}.jar
+          modules/transformer-pubsub/target/scala-2.12/snowplow-transformer-pubsub-${{ steps.ver.outputs.project_version }}.jar
       env:
         GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

README.md

@@ -8,29 +8,38 @@
 
 ## Introduction
 
-This project contains applications required to load Snowplow data into relational databases.
+This project contains applications required to load Snowplow data into various data warehouses.
 
-### RDB Shredder
+It consists of two types of applications: Transformers and Loaders
 
-RDB Shredder is a [Spark][spark] job which:
+### Transformers
 
-1. Reads Snowplow enriched events from S3
-2. Extracts any unstructured event JSONs and context JSONs found
-3. Validates that these JSONs conform to schema
-4. Adds metadata to these JSONs to track their origins
-5. Writes these JSONs out to nested folders dependent on their schema
+Transformers read Snowplow enriched events, transform them to a format ready to be loaded to a data warehouse, then write them to respective blob storage.
 
-It is designed to be run downstream of the [Enrich][enrich] job.
+There are two types of Transformers: Batch and Streaming
 
-### RDB Loader
+#### Stream Transformer
 
-RDB Loader (previously known as StorageLoader) is a Scala application that runs in background, discovering [data][shred], produced by RDB Shredder from SQS queue and loading it into one of possible [storage targets][targets].
+Stream Transformers read enriched events from respective stream service, transform them, then write transformed events to specified blob storage path.
+They write transformed events in periodic windows.
 
-### RDB Stream Shredder (experimental)
+There are two different Stream Transformer applications: Transformer Kinesis and Transformer Pubsub. As one can predict, they are different variants for GCP and AWS.
 
-An application similar to RDB Shredder, but working without Apache Spark or EMR
-and reading directly from Kinesis Stream. Only Shredder or Stream Shredder
-should be used.
+
+#### Batch Transformer
+
+It is a [Spark][spark] job. It only works with AWS services. It reads enriched events from a given S3 path, transforms them, then writes transformed events to a specified S3 path.
+
+
+### Loaders
+
+Transformers send a message to a message queue after they are finished with transforming some batch and writing it to blob storage.
+This message contains information about transformed data such as where it is stored and what it looks like.
+
+Loaders subscribe to the message queue. After a message is received, it is parsed, and necessary bits are extracted to load transformed events to the destination.
+Loaders construct necessary SQL statements to load transformed events then they send these SQL statements to the specified destination.
+
+At the moment, we have loader applications for Redshift, Databricks and Snowflake.
 
 ## Find out more
 
@@ -41,7 +50,7 @@ should be used.
 
 ## Copyright and License
 
-Snowplow Relational Database Loader is copyright 2012-2021 Snowplow Analytics Ltd.
+Snowplow Relational Database Loader is copyright 2012-2022 Snowplow Analytics Ltd.
 
 Licensed under the **[Apache License, Version 2.0][license]** (the "License");
 you may not use this software except in compliance with the License.
@@ -56,15 +65,11 @@ limitations under the License.
 [techdocs-image]: https://d3i6fms1cm1j0i.cloudfront.net/github/images/techdocs.png
 [setup-image]: https://d3i6fms1cm1j0i.cloudfront.net/github/images/setup.png
 [roadmap-image]: https://d3i6fms1cm1j0i.cloudfront.net/github/images/roadmap.png
-[setup]: https://docs.snowplowanalytics.com/docs/getting-started-on-snowplow-open-source/setup-snowplow-on-aws/setup-destinations/setup-redshift/
-[techdocs]: https://docs.snowplowanalytics.com/docs/pipeline-components-and-applications/loaders-storage-targets/snowplow-rdb-loader/
+[setup]: https://docs.snowplow.io/docs/getting-started-on-snowplow-open-source/
+[techdocs]: https://docs.snowplow.io/docs/pipeline-components-and-applications/loaders-storage-targets/snowplow-rdb-loader/
 [roadmap]: https://github.com/snowplow/snowplow/projects/7
 
 [spark]: http://spark.apache.org/
-[enrich]: https://github.com/snowplow/snowplow/enrich
-
-[targets]: https://github.com/snowplow/snowplow/wiki/Configuring-storage-targets
-[shred]: https://docs.snowplowanalytics.com/docs/pipeline-components-and-applications/loaders-storage-targets/snowplow-rdb-loader/rdb-shredder/
 
 [build-image]: https://github.com/snowplow/snowplow-rdb-loader/workflows/Test%20and%20deploy/badge.svg
 [build]: https://github.com/snowplow/snowplow-rdb-loader/actions?query=workflow%3A%22Test%22

build.sbt

@@ -12,20 +12,50 @@
  */
 
 lazy val root = project.in(file("."))
-  .aggregate(common, aws, loader, databricksLoader, redshiftLoader, snowflakeLoader, transformerBatch, transformerKinesis)
+  .aggregate(
+    aws,
+    gcp,
+    common,
+    commonTransformerStream,
+    loader,
+    databricksLoader,
+    redshiftLoader,
+    snowflakeLoader,
+    transformerBatch,
+    transformerKinesis,
+    transformerPubsub,
+  )
 
-lazy val aws = project
+lazy val common: Project = project
+  .in(file("modules/common"))
+  .settings(BuildSettings.commonBuildSettings)
+  .settings(libraryDependencies ++= Dependencies.commonDependencies)
+  .settings(excludeDependencies ++= Dependencies.exclusions)
+  .enablePlugins(BuildInfoPlugin)
+
+lazy val aws: Project = project
   .in(file("modules/aws"))
   .settings(BuildSettings.awsBuildSettings)
-  .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
   .settings(libraryDependencies ++= Dependencies.awsDependencies)
+  .settings(excludeDependencies ++= Dependencies.exclusions)
+  .dependsOn(common % "compile->compile;test->test")
   .enablePlugins(BuildInfoPlugin)
 
-lazy val common: Project = project
-  .in(file("modules/common"))
-  .settings(BuildSettings.commonBuildSettings)
-  .settings(libraryDependencies ++= Dependencies.commonDependencies)
+lazy val gcp: Project = project
+  .in(file("modules/gcp"))
+  .settings(BuildSettings.gcpBuildSettings)
+  .settings(libraryDependencies ++= Dependencies.gcpDependencies)
   .settings(excludeDependencies ++= Dependencies.exclusions)
+  .dependsOn(common % "compile->compile;test->test")
+  .enablePlugins(BuildInfoPlugin)
+
+lazy val commonTransformerStream = project
+  .in(file("modules/common-transformer-stream"))
+  .settings(BuildSettings.commonStreamTransformerBuildSettings)
+  .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
+  .settings(libraryDependencies ++= Dependencies.commonStreamTransformerDependencies)
+  .settings(excludeDependencies ++= Dependencies.commonStreamTransformerExclusions)
+  .dependsOn(common % "compile->compile;test->test")
   .enablePlugins(BuildInfoPlugin)
 
 lazy val loader = project
@@ -34,15 +64,15 @@ lazy val loader = project
   .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
   .settings(libraryDependencies ++= Dependencies.loaderDependencies)
   .settings(excludeDependencies ++= Dependencies.exclusions)
-  .dependsOn(common % "compile->compile;test->test", aws)
+  .dependsOn(aws % "compile->compile;test->test", gcp % "compile->compile;test->test")
   .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin)
 
 lazy val redshiftLoader = project
   .in(file("modules/redshift-loader"))
   .settings(BuildSettings.redshiftBuildSettings)
   .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
   .settings(libraryDependencies ++= Dependencies.redshiftDependencies)
-  .dependsOn(common % "compile->compile;test->test", aws, loader % "compile->compile;test->test")
+  .dependsOn(loader % "compile->compile;test->test")
   .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin)
 
 lazy val redshiftLoaderDistroless = project
@@ -51,15 +81,15 @@ lazy val redshiftLoaderDistroless = project
   .settings(BuildSettings.redshiftDistrolessBuildSettings)
   .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
   .settings(libraryDependencies ++= Dependencies.redshiftDependencies)
-  .dependsOn(common % "compile->compile;test->test", aws, loader % "compile->compile;test->test")
+  .dependsOn(loader % "compile->compile;test->test")
   .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin, LauncherJarPlugin)
 
 lazy val snowflakeLoader = project
   .in(file("modules/snowflake-loader"))
   .settings(BuildSettings.snowflakeBuildSettings)
   .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
   .settings(libraryDependencies ++= Dependencies.snowflakeDependencies)
-  .dependsOn(common % "compile->compile;test->test", aws, loader % "compile->compile;test->test")
+  .dependsOn(common % "compile->compile;test->test",loader % "compile->compile;test->test")
   .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin)
 
 lazy val snowflakeLoaderDistroless = project
@@ -68,22 +98,22 @@ lazy val snowflakeLoaderDistroless = project
   .settings(BuildSettings.snowflakeDistrolessBuildSettings)
   .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
   .settings(libraryDependencies ++= Dependencies.snowflakeDependencies)
-  .dependsOn(common % "compile->compile;test->test", aws, loader % "compile->compile;test->test")
+  .dependsOn(loader % "compile->compile;test->test")
   .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin, LauncherJarPlugin)
 
 lazy val databricksLoader = project
   .in(file("modules/databricks-loader"))
   .settings(BuildSettings.databricksBuildSettings)
   .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
-  .dependsOn(common % "compile->compile;test->test", aws, loader % "compile->compile;test->test")
+  .dependsOn(loader % "compile->compile;test->test")
   .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin)
 
 lazy val databricksLoaderDistroless = project
   .in(file("modules/distroless/databricks-loader"))
   .settings(sourceDirectory := (databricksLoader / sourceDirectory).value)
   .settings(BuildSettings.databricksDistrolessBuildSettings)
   .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
-  .dependsOn(common % "compile->compile;test->test", aws, loader % "compile->compile;test->test")
+  .dependsOn(loader % "compile->compile;test->test")
   .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin, LauncherJarPlugin)
 
 lazy val transformerBatch = project
@@ -98,17 +128,28 @@ lazy val transformerKinesis = project
   .in(file("modules/transformer-kinesis"))
   .settings(BuildSettings.transformerKinesisBuildSettings)
   .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
-  .settings(libraryDependencies ++= Dependencies.transformerKinesisDependencies)
-  .settings(excludeDependencies ++= Dependencies.transformerKinesisExclusions)
-  .dependsOn(common % "compile->compile;test->test", aws)
+  .dependsOn(commonTransformerStream % "compile->compile;test->test", aws % "compile->compile;test->test")
   .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin)
 
 lazy val transformerKinesisDistroless = project
   .in(file("modules/distroless/transformer-kinesis"))
   .settings(sourceDirectory := (transformerKinesis / sourceDirectory).value)
   .settings(BuildSettings.transformerKinesisDistrolessBuildSettings)
   .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
-  .settings(libraryDependencies ++= Dependencies.transformerKinesisDependencies)
-  .settings(excludeDependencies ++= Dependencies.transformerKinesisExclusions)
-  .dependsOn(common, aws)
+  .dependsOn(commonTransformerStream, aws)
+  .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin, LauncherJarPlugin)
+
+lazy val transformerPubsub = project
+  .in(file("modules/transformer-pubsub"))
+  .settings(BuildSettings.transformerPubsubBuildSettings)
+  .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
+  .dependsOn(commonTransformerStream % "compile->compile;test->test", gcp % "compile->compile;test->test")
+  .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin)
+
+lazy val transformerPubsubDistroless = project
+  .in(file("modules/distroless/transformer-pubsub"))
+  .settings(sourceDirectory := (transformerPubsub / sourceDirectory).value)
+  .settings(BuildSettings.transformerPubsubDistrolessBuildSettings)
+  .settings(addCompilerPlugin(Dependencies.betterMonadicFor))
+  .dependsOn(commonTransformerStream, gcp)
   .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin, LauncherJarPlugin)

config/databricks.config.minimal.hocon renamed to config/loader/aws/databricks.config.minimal.hocon

@@ -17,9 +17,8 @@
     "host": "abc.cloud.databricks.com",
     # DB password
     "password": {
-      # A password can be placed in EC2 parameter store or be a plain text
-      "ec2ParameterStore": {
-        # EC2 parameter name
+      # A password can be placed in EC2 parameter store, GCP Secret Manager or be a plain text
+      "secretStore": {
         "parameterName": "snowplow.databricks.password"
       }
     },

config/databricks.config.reference.hocon renamed to config/loader/aws/databricks.config.reference.hocon

@@ -1,4 +1,6 @@
 {
+  # Specifies the cloud provider that application will be deployed into
+  "cloud": "aws"
   # Data Lake (S3) region
   # This field is optional if it can be resolved with AWS region provider chain.
   # It checks places like env variables, system properties, AWS profile file.

config/redshift.config.minimal.hocon renamed to config/loader/aws/redshift.config.minimal.hocon

@@ -22,9 +22,8 @@
     "username": "admin",
     # DB password
     "password": {
-      # A password can be placed in EC2 parameter store or be a plain text
-      "ec2ParameterStore": {
-        # EC2 parameter name
+      # A password can be placed in EC2 parameter store, GCP Secret Manager or be a plain text
+      "secretStore": {
         "parameterName": "snowplow.snowflake.password"
       }
     },

config/redshift.config.reference.hocon renamed to config/loader/aws/redshift.config.reference.hocon

@@ -0,0 +1,22 @@
+{
+  "messageQueue": {
+    "type": "pubsub"
+    "subscription": "projects/project-id/subscriptions/subscription-id"
+  },
+  "storage" : {
+    "type": "snowflake",
+
+    "snowflakeRegion": "us-west-2",
+    "username": "admin",
+    "password": "Supersecret1",
+    "account": "acme",
+    "warehouse": "wh",
+    "schema": "atomic",
+    "database": "snowplow",
+
+    "transformedStage": {
+      "name": "snowplow_stage"
+      "location": "gs://bucket/transformed/"
+    }
+  }
+}