Update all plans to use connection builders, flatten structure to make it clear what data source is used

Author: pflooky

.gitignore

@@ -31,6 +31,7 @@ docker/data/custom/generated
 docker/data/custom/recordTracking
 docker/data/custom/report
 docker/sample
+docker/tmp
 
 benchmark/jars
 

Dockerfile

@@ -1,3 +1,3 @@
-FROM datacatering/data-caterer:0.13.1
+FROM datacatering/data-caterer:0.14.0
 
 COPY --chown=app:app build/libs/data-caterer-example-0.1.0.jar /opt/app/job.jar

docker/data/custom/task/http/http-account-task-simple.yaml

@@ -0,0 +1,71 @@
+name: "simple_http"
+steps:
+  - name: "account"
+    count:
+      records: 50
+    fields:
+      - name: "httpUrl"
+        fields:
+          - name: "url"
+            static: "http://localhost:80/anything/{id}"
+          - name: "method"
+            static: "PUT"
+          - name: "pathParam"
+            fields:
+              - name: "id"
+                options:
+                  sql: "bodyContent.account_id"
+          - name: "queryParam"
+            fields:
+              - name: "limit"
+                type: "integer"
+                options:
+                  oneOf:
+                    - 10
+                    - 5
+      - name: "httpHeaders"
+        fields:
+          - name: "Content-Type"
+            static: "application/json"
+          - name: "X-Account-Id"
+            options:
+              sql: "bodyContent.account_id"
+          - name: "X-Updated"
+            type: "timestamp"
+            options:
+              sql: "bodyContent.details.updated_by.time"
+      - name: "httpBody"
+        fields:
+          - name: "account_id"
+            options:
+              regex: "ACC[0-9]{8}"
+          - name: "year"
+            type: "int"
+            options:
+              min: 2021
+              max: 2022
+          - name: "amount"
+            type: "double"
+            options:
+              min: 10.0
+              max: 100.0
+          - name: "details"
+            fields:
+              - name: "name"
+              - name: "txn_date"
+                type: "date"
+                options:
+                  min: "2021-01-01"
+                  max: "2021-12-31"
+              - name: "updated_by"
+                fields:
+                  - name: "user"
+                  - name: "time"
+                    type: "timestamp"
+          - name: "transactions"
+            type: "array"
+            fields:
+              - name: "txn_date"
+                type: "date"
+              - name: "amount"
+                type: "double"

docker/data/custom/task/kafka/kafka-account-task-simple.yaml

@@ -0,0 +1,54 @@
+name: "simple_kafka"
+steps:
+  - name: "kafka_account"
+    type: "json"
+    count:
+      records: "10"
+    options:
+      topic: "account-topic"
+    fields:
+      - name: "key"
+        type: "string"
+        options:
+          sql: "content.account_id"
+      - name: "messageBody"
+        fields:
+          - name: "account_id"
+          - name: "year"
+            type: "int"
+            options:
+              min: "2021"
+              max: "2022"
+          - name: "amount"
+            type: "double"
+            options:
+              min: "10.0"
+              max: "100.0"
+          - name: "details"
+            fields:
+              - name: "name"
+              - name: "txn_date"
+                type: "date"
+                options:
+                  min: "2021-01-01"
+                  max: "2021-12-31"
+              - name: "updated_by"
+                fields:
+                  - name: "user"
+                  - name: "time"
+                    type: "timestamp"
+          - name: "transactions"
+            type: "array"
+            fields:
+              - name: "txn_date"
+                type: "date"
+              - name: "amount"
+                type: "double"
+      - name: "messageHeaders"
+        fields:
+          - name: "account-id"
+            options:
+              sql: "content.account_id"
+          - name: "updated"
+            options:
+              sql: "content.details.update_by.time"

gradle.properties

@@ -8,5 +8,5 @@ version=0.1.0
 
 scalaVersion=2.12
 scalaSpecificVersion=2.12.19
-dataCatererVersion=0.13.1
+dataCatererVersion=0.14.1
 sparkMajorVersion=3.5

helm/data-caterer/values.yaml

@@ -8,7 +8,7 @@ image:
   repository: "datacatering/data-caterer"
   pullPolicy: "IfNotPresent"
   # Overrides the image tag whose default is the chart appVersion.
-  tag: "0.13.1"
+  tag: "0.14.0"
 
 imagePullSecrets: []
 nameOverride: ""

run.sh

@@ -1,6 +1,20 @@
 #!/usr/bin/env bash
 
 data_caterer_version=$(grep dataCatererVersion gradle.properties | cut -d= -f2)
+data_caterer_user=${DATA_CATERER_API_USER:-}
+data_caterer_token=${DATA_CATERER_API_TOKEN:-}
+
+echo "Checking for Data Caterer user and token..."
+if [[ -z ${DATA_CATERER_API_USER} ]]; then
+  read -p "Data Caterer user: " data_caterer_user
+  DATA_CATERER_API_USER=data_caterer_user
+fi
+if [[ -z ${DATA_CATERER_API_TOKEN} ]]; then
+  read -p "Data Caterer token: " -s data_caterer_token
+  DATA_CATERER_API_TOKEN=data_caterer_token
+  echo
+fi
+
 if [[ -s ".tmp_prev_class_name" ]]; then
   prev_class_name=$(cat .tmp_prev_class_name)
 else
@@ -38,11 +52,14 @@ DOCKER_CMD=(
   -v "$(pwd)/docker/sample/tracking:/opt/app/record-tracking"
   -v "$(pwd)/docker/mount:/opt/app/mount"
   -v "$(pwd)/docker/data/custom:/opt/app/custom"
+  -v "$(pwd)/docker/tmp:/tmp"
   -e "APPLICATION_CONFIG_PATH=/opt/app/custom/application.conf"
   -e "$full_class_name"
   -e "DEPLOY_MODE=client"
   -e "DRIVER_MEMORY=2g"
   -e "EXECUTOR_MEMORY=2g"
+  -e "DATA_CATERER_API_USER=$data_caterer_user"
+  -e "DATA_CATERER_API_TOKEN=$data_caterer_token"
   --network "docker_default"
   datacatering/data-caterer:"$data_caterer_version"
 )

src/main/scala/io/github/datacatering/plan/AdvancedBatchEventPlanRun.scala

@@ -4,7 +4,7 @@ import io.github.datacatering.datacaterer.api.PlanRun
 
 class AdvancedBatchEventPlanRun extends PlanRun {
 
-  val kafkaTask = new AdvancedKafkaPlanRun().kafkaTask
+  val kafkaTask = new KafkaPlanRun().kafkaTask
 
   val csvTask = csv("my_csv", "/opt/app/data/csv/account")
     .fields(

src/main/scala/io/github/datacatering/plan/AdvancedKafkaPlanRun.scala

@@ -11,63 +11,48 @@ class AdvancedPlanRun extends PlanRun {
   val accountIdField = field.name("account_id").regex("ACC[0-9]{8}")
   val nameField = field.name("name").expression("#{Name.name}")
 
-  val postgresTask = task.name("customer_postgres")
-    .steps(
-      step
-        .name("transaction")
-        .jdbcTable("account.transaction")
-        .fields(
-          accountIdField,
-          field.name("txn_id").regex("txn_[0-9]{5}"),
-          field.name("year").`type`(IntegerType).sql("YEAR(date)"),
-          nameField,
-          field.name("date").`type`(DateType).min(startDate),
-          field.name("amount").`type`(DoubleType).max(10000),
-          field.name("credit_debit").sql("CASE WHEN amount < 0 THEN 'C' ELSE 'D' END"),
-        )
-        .count(
-          count
-            .recordsPerFieldGenerator(generator.min(1).max(10), "account_id")
-        )
-      ,
-      step
-        .name("account")
-        .jdbcTable("account.account")
-        .fields(
-          accountIdField,
-          nameField,
-          field.name("open_date").`type`(DateType).min(startDate),
-          field.name("status").oneOf("open", "closed", "pending")
-        )
-        .count(10)
+  val postgresTxn = postgres("customer_postgres_txn")
+    .table("account.transactions")
+    .fields(
+      accountIdField,
+      field.name("txn_id").regex("txn_[0-9]{5}"),
+      field.name("year").`type`(IntegerType).sql("YEAR(date)"),
+      nameField,
+      field.name("date").`type`(DateType).min(startDate),
+      field.name("amount").`type`(DoubleType).max(10000),
+      field.name("credit_debit").sql("CASE WHEN amount < 0 THEN 'C' ELSE 'D' END"),
+    )
+    .count(
+      count
+        .recordsPerFieldGenerator(generator.min(1).max(10), "account_id")
     )
+  val postgresAccount = postgres("customer_postgres_account")
+    .table("account.account")
+    .fields(
+      accountIdField,
+      nameField,
+      field.name("open_date").`type`(DateType).min(startDate),
+      field.name("status").oneOf("open", "closed", "pending")
+    )
+    .count(count.records(10))
 
-  val jsonTask = task.name("account_json")
-    .steps(
-      step
-        .name("account_info")
-        .path("src/main/resources/sample/json")
+  val jsonTask = json("account_json", "src/main/resources/sample/json")
+    .fields(
+      accountIdField,
+      nameField,
+      field.name("txn_list")
+        .`type`(ArrayType)
         .fields(
-          accountIdField,
-          nameField,
-          field.name("txn_list")
-            .`type`(ArrayType)
-            .fields(
-              field.name("id").sql("_holding_txn_id"),
-              field.name("date").`type`(DateType).min(startDate),
-              field.name("amount").`type`(DoubleType),
-            ),
-          field.name("_holding_txn_id").omit(true)
-        )
+          field.name("id").sql("_holding_txn_id"),
+          field.name("date").`type`(DateType).min(startDate),
+          field.name("amount").`type`(DoubleType),
+        ),
+      field.name("_holding_txn_id").omit(true)
     )
 
   val conf = configuration.postgres("my_postgres")
 
   val accountPlan = plan.name("Create accounts and transactions across Postgres and JSON file")
-    .taskSummaries(
-      taskSummary.dataSource("my_postgres").task(postgresTask),
-      taskSummary.dataSource("my_json").task(jsonTask)
-    )
     .addForeignKeyRelationship(
       foreignField("my_postgres", "transaction", "txn_id"),
       foreignField("my_json", "account_info", "_holding_txn_id")
@@ -83,5 +68,5 @@ class AdvancedPlanRun extends PlanRun {
       foreignField("my_json", "account_info", "name"),
     )
 
-  execute(accountPlan, conf)
+  execute(accountPlan, conf, postgresAccount, postgresTxn, jsonTask)
 }