A step by step guide to monitor spark jobs running in K8 via Prometheus

Set up the Prometheus server using kube-prometheus-stack helm chart on Docker-On-Mac

Please follow my post to set up the Prometheus server on Docker-On-Mac

Prometheus on Docker-On-Mac

Inspect the Prometheus resource created by the kube-prometheus-stack helm chart

A Prometheus, defines a desired state of Prometheus deployment.

The Prometheus created by kube-prometheus-stack is looking for a service monitor objects defined with label: `release: prometheus`

Lets’ check the Prometheus resource. Execute below command:

kubectl get prometheus -oyaml

The yaml file looks like below and check for `serviceMonitorSelector`

apiVersion: v1
items:
- apiVersion: monitoring.coreos.com/v1
  kind: Prometheus
  metadata:
    annotations:
      meta.helm.sh/release-name: prometheus
      meta.helm.sh/release-namespace: default
    creationTimestamp: "2021-02-24T20:12:37Z"
    generation: 1
    labels:
      app: kube-prometheus-stack-prometheus
      app.kubernetes.io/managed-by: Helm
      chart: kube-prometheus-stack-13.10.0
      heritage: Helm
      release: prometheus
    managedFields:
    - apiVersion: monitoring.coreos.com/v1
      fieldsType: FieldsV1
      fieldsV1:
        f:metadata:
          f:annotations:
            .: {}
            f:meta.helm.sh/release-name: {}
            f:meta.helm.sh/release-namespace: {}
          f:labels:
            .: {}
            f:app: {}
            f:app.kubernetes.io/managed-by: {}
            f:chart: {}
            f:heritage: {}
            f:release: {}
        f:spec:
          .: {}
          f:alerting:
            .: {}
            f:alertmanagers: {}
          f:enableAdminAPI: {}
          f:externalUrl: {}
          f:image: {}
          f:listenLocal: {}
          f:logFormat: {}
          f:logLevel: {}
          f:paused: {}
          f:podMonitorNamespaceSelector: {}
          f:podMonitorSelector:
            .: {}
            f:matchLabels:
              .: {}
              f:release: {}
          f:portName: {}
          f:probeNamespaceSelector: {}
          f:probeSelector:
            .: {}
            f:matchLabels:
              .: {}
              f:release: {}
          f:replicas: {}
          f:retention: {}
          f:routePrefix: {}
          f:ruleNamespaceSelector: {}
          f:ruleSelector:
            .: {}
            f:matchLabels:
              .: {}
              f:app: {}
              f:release: {}
          f:securityContext:
            .: {}
            f:fsGroup: {}
            f:runAsGroup: {}
            f:runAsNonRoot: {}
            f:runAsUser: {}
          f:serviceAccountName: {}
          f:serviceMonitorNamespaceSelector: {}
          f:serviceMonitorSelector:
            .: {}
            f:matchLabels:
              .: {}
              f:release: {}
          f:shards: {}
          f:version: {}
      manager: Go-http-client
      operation: Update
      time: "2021-02-24T20:12:37Z"
    name: prometheus-kube-prometheus-prometheus
    namespace: default
    resourceVersion: "1383"
    selfLink: /apis/monitoring.coreos.com/v1/namespaces/default/prometheuses/prometheus-kube-prometheus-prometheus
    uid: 1315fc95-3465-45a0-bcbf-1ce924252925
  spec:
    alerting:
      alertmanagers:
      - apiVersion: v2
        name: prometheus-kube-prometheus-alertmanager
        namespace: default
        pathPrefix: /
        port: web
    enableAdminAPI: false
    externalUrl: http://prometheus-kube-prometheus-prometheus.default:9090
    image: quay.io/prometheus/prometheus:v2.24.0
    listenLocal: false
    logFormat: logfmt
    logLevel: info
    paused: false
    podMonitorNamespaceSelector: {}
    podMonitorSelector:
      matchLabels:
        release: prometheus
    portName: web
    probeNamespaceSelector: {}
    probeSelector:
      matchLabels:
        release: prometheus
    replicas: 1
    retention: 10d
    routePrefix: /
    ruleNamespaceSelector: {}
    ruleSelector:
      matchLabels:
        app: kube-prometheus-stack
        release: prometheus
    securityContext:
      fsGroup: 2000
      runAsGroup: 2000
      runAsNonRoot: true
      runAsUser: 1000
    serviceAccountName: prometheus-kube-prometheus-prometheus
    serviceMonitorNamespaceSelector: {}
    serviceMonitorSelector:
      matchLabels:
        release: prometheus
    shards: 1
    version: v2.24.0
kind: List
metadata:
  resourceVersion: ""
  selfLink: ""

Create a Prometheus Service Monitor that is listening to a Service with a specific label

ServiceMonitor, which declaratively specifies how groups of Kubernetes services should be monitored. The Operator automatically generates Prometheus scrape configuration based on the current state of the objects in the API server.

The Prometheus resource is looking for a service monitor resources with label `release: prometheus` so we need to ensure that this label is defined in our service monitor resource.

Let’s use the below yaml file to create a service monitoring resource that is listening to services defined with label `app: spark`

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: servicemonitor-spark
  labels:
    release: prometheus
spec:
  selector:
    matchLabels:
      app: spark
  endpoints:
    - port: metrics

Save the file as servicemonitor-spark.yaml and save it inside a folder name prometheus and execute following:

kubectl apply -f prometheus/servicemonitor-spark.yaml

Check the object via

kubectl get servicemonitor

Create a service resource listening to pods with a specific label

The service monitor resource is looking for a service resources with label `app: spark` so we need to ensure that this label is defined in our service resource.

The service resource looks for the pods with label `app: spark` and the pods should be exposing their metrics at port 8090

Save the content below in a file say spark-service.yaml and save it inside a folder prometheus.

apiVersion: v1
kind: Service
metadata:
  name: spark-service
  labels:
    app: spark
spec:
  ports:
    - name: metrics
      port: 8090
      targetPort: 8090
      protocol: TCP
  selector:
    app: spark

And execute

kubectl apply -f prometheus/spark-service.yaml

Run a sample Spark Job in K8

Please go through my earlier post to set up the spark-k8-operator

Run Spark Job on Kubernetes using Spark-on-k8s-Operator

Let’s use the below yam file to deploy a sample Spark job.

The driver and executor defines the label `app: spark` which is required by the service resource.

Save the file with name spark-sample-prometheus.yaml, inside a folder name prometheus.

apiVersion: "sparkoperator.k8s.io/v1beta2"
kind: ScheduledSparkApplication
metadata:
  name: spark-pi-scheduled
  namespace: default
spec:
  schedule: "@every 1m"
  concurrencyPolicy: Allow
  template:
    type: Scala
    mode: cluster
    image: "gcr.io/spark-operator/spark:v3.0.0-gcs-prometheus"
    imagePullPolicy: Always
    mainClass: org.apache.spark.examples.SparkPi
    mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.12-3.0.0.jar"
    sparkVersion: "3.0.0"
    restartPolicy:
      type: Never
    driver:
      cores: 1
      coreLimit: "1200m"
      memory: "512m"
      labels:
        version: 3.0.0
      serviceAccount: spark
    executor:
      cores: 1
      instances: 1
      memory: "512m"
      labels:
        version: 3.0.0
    monitoring:
      exposeDriverMetrics: true
      exposeExecutorMetrics: true
      prometheus:
        jmxExporterJar: "/prometheus/jmx_prometheus_javaagent-0.11.0.jar"
        port: 8090

Execute the below command to deploy the spark job:

kubectl apply -f prometheus/spark-sample-prometheus.yaml

The metrics would be accessible at port 8090

View the metrics in prometheus server

Please go through my earlier post to set up the K8 and K8 Dashboard on Docker-On-Mac

Running Kubernetes Dashboard On Docker-On-Mac

Install the helm chart

We will use kube-prometheus-stack helm chart for getting Prometheus server running on our Docker-On-Mac.

Add the repo

helm repo add prometheus-community https://prometheus-community.github.io/helm-charts

Update the repo

helm repo update

Install the chart

helm install prometheus prometheus-community/kube-prometheus-stack

Check the pods

kubectl get pods

Check the services

kubectl get svc

Expose the Prometheus and Grafana servers from localhost

Save the below manifest file in a file say prometheus-service.yaml and save it inside a folder name prometheus.

apiVersion: v1
kind: Service
metadata:
  name: prometheus
spec:
  type: NodePort
  ports:
    - name: web
      nodePort: 30900
      port: 9090
      protocol: TCP
      targetPort: web
  selector:
    prometheus: prometheus-kube-prometheus-prometheus

---
apiVersion: v1
kind: Service
metadata:
  name: grafana
spec:
  type: NodePort
  ports:
    - name: service
      nodePort: 30901
      port: 80
      protocol: TCP
      targetPort: 3000
  selector:
    app.kubernetes.io/name: grafana

Execute:

kubectl apply -f prometheus/prometheus-service.yaml

The prometheus server can now be accessed via http://localhost:30900/ and grafana at http://localhost:30901/?orgId=1

Grafana username/password is admin/prom-operator

Check out my another post on how to create a custom resource monitors

https://suchit-g.medium.com/spark-with-prometheus-monitoring-2f91e1208eed

Objective

Deploy Zeppelin:0.9.0 in K8

Authenticate Zeppelin UI via Apache Shiro JDBC realm

Repo

Code being discussed can be found in Github

Steps to meet the objective

Enable Kubernetes

Follow below link to run K8 on Mac.

Running Kubernetes Dashboard On Docker-On-Mac

Build the docker image

docker build . -t sg/zeppelin:latest

Dockerfile

#https://hub.docker.com/r/apache/zeppelin/dockerfile
ARG ZEPPELIN_IMAGE=apache/zeppelin:0.9.0
FROM ${ZEPPELIN_IMAGE}

WORKDIR ${Z_HOME}

USER root
ADD /shiro.ini ${Z_HOME}/conf/
#Required for local set up. Overriding so as to make imagePullPolicy as Never
ADD /100-interpreter-spec.yaml ${Z_HOME}/k8s/interpreter/

#Once ZEPPELIN-5192 is resolved, we can remove the below command. 
ADD /credentials.json ${Z_HOME}/conf/

#Required for starting the Zeppelin server. Without this Shiro won't be able to set the env configuration.
ADD https://repo1.maven.org/maven2/org/postgresql/postgresql/42.2.5/postgresql-42.2.5.jar  ${Z_HOME}/lib
ENV CLASSPATH=${Z_HOME}/lib/postgresql-42.2.5.jar:${CLASSPATH}
RUN chmod 775 ${Z_HOME}/lib/postgresql-42.2.5.jar

USER 1000

ENTRYPOINT [ "/usr/bin/tini", "--" ]
WORKDIR ${Z_HOME}
CMD ["bin/zeppelin.sh"]

Why we need a custom image?

We need to provide Zeppelin UI Authentication via Apache Shiro. So need to include shiro.ini as part of the docker image.

Why we need to override 100-interpreter-spec.yaml?

Since the docker image sg/zeppelin:latest is not in docker hub, we need to override the file to make imagePolicy as never

Why credentials.json is part of the docker image?

This should get resolved with ZEPPELIN-5192

Build the stack for Kubernetes

This will build the postgresql image with Zeppelin database with following tables:

1. zeppelin.zeppelin_users
2. zeppelin.zeppelin_user_roles

The tables will populated with user zeppelin_ui_user and encrypted password Welcome123

docker-compose -f docker-compose-zeppelin.yaml build

Push the stack to kubernetes

docker stack deploy --compose-file docker-compose-zeppelin.yaml  --orchestrator kubernetes zeppelin

You can connect the database with zeppelin_user/Password123

Pre populated user

PostgreSQL deployed in K8

Apply the kubernetes manifest file

kubectl apply -f zeppelin-server-manifest.yaml

Why custom manifest file?

1. ZEPPELIN-5191: The manifest provided by Zeppelin is pointing to wrong docker image.
2. We need to use a custom docker image.

Open the proxy terminal for Zeppelin UI

Open a terminal and execute

kubectl port-forward deployment/zeppelin-server 8080:8080

Access the Zeppelin UI

Go to http://localhost:8080/ and try zeppelin_ui_user/Welcome123

Creating a new user

Generate Password

Execute `java -jar shiro-tools-hasher-1.5.0-cli.jar -p` and enter the password that you want to hash.

Insert the user in zeppelin.zeppelin_users table

INSERT INTO zeppelin.zeppelin_users (id, username, password, created_at, updated_at) VALUES (DEFAULT, 'test_ui', '$shiro1$SHA-256$500000$8FQTUt1U4IHPv3CZgs2k3A==$n8ZSN4zc6PJ4fMs9el6i1sOVObCez0mryQ5tNjFZ2TA=', DEFAULT, DEFAULT);

Assign a role to the user.

INSERT INTO zeppelin.zeppelin_user_roles (id, username, role_name, created_at, updated_at) VALUES (DEFAULT, 'test_ui', 'admin', DEFAULT, DEFAULT);

Hope this was useful :)

Set up:

1. Docker on Mac: 2.2.0
2. Kubernetes: 1.15.5
3. macOS: 10.15.6
4. Spark: 3.0.0 with hadoop 3.2
5. Spark Google operator
6. minIO: latest image

Step 1: Enable K8 and set up Dashboard on local.

Running Kubernetes Dashboard On Docker-On-Mac

Step 2: Set up the spark google operator

Run Spark Job on Kubernetes using Spark-on-k8s-Operator

Step 3: Get the minIO in the docker-compose and push the stack

docker-compose.yaml

version: '3.0'
services:
  minio:
    image: minio/minio
    container_name: minio
    ports:
      - 9000:9000
    environment:
      MINIO_ACCESS_KEY: AKIAIOSFODNN7EXAMPLE
      MINIO_SECRET_KEY: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
    entrypoint: sh
    command: -c 'mkdir -p /export/spark-events && mkdir -p /export/spark-events/logs && /usr/bin/minio server /export'
    healthcheck:
      test: [ "CMD", "curl", "-f", "http://localhost:9000/minio/health/live" ]
      interval: 30s
      timeout: 20s
      retries: 3

Command to push the stack:

docker stack deploy --compose-file docker-compose-local-setup.yaml --namespace default --orchestrator kubernetes spark-minio 

After the push, UI should be accessible at http://localhost:9000/minio/

Step 4: Set up the spark history server

Dockerfile:

ARG SPARK_IMAGE=gcr.io/spark-operator/spark:v3.0.0-hadoop3
FROM ${SPARK_IMAGE}
# Switch to user root so we can add additional jars, packages and configuration files.
USER root

RUN apt-get -y update && apt-get install -y coreutils

RUN rm -rf $SPARK_HOME/jars/httpclient-4.5.6.jar

# Add dependency for hadoop-aws
ADD https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-core/1.11.883/aws-java-sdk-core-1.11.883.jar $SPARK_HOME/jars
ADD https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-dynamodb/1.11.883/aws-java-sdk-dynamodb-1.11.883.jar $SPARK_HOME/jars
ADD https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-s3/1.11.883/aws-java-sdk-s3-1.11.883.jar $SPARK_HOME/jars

ADD https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.2.0/hadoop-aws-3.2.0.jar $SPARK_HOME/jars

ADD https://repo1.maven.org/maven2/org/apache/httpcomponents/httpclient/4.5.9/httpclient-4.5.9.jar $SPARK_HOME/jars

#Required for minIO service discovery by name.
ADD /spark-history/spark-defaults.conf /opt/spark/conf/

ENTRYPOINT ["/opt/entrypoint.sh"]

spark-defaults.conf

spark.hadoop.fs.s3a.path.style.access  true

Build the docker image

docker build -f spark-history/Dockerfile -t spark-history-server:latest .

Create secret for S3

kubectl create secret generic aws-secrets --from-file=secret/aws-access-key --from-file=secret/aws-secret-key

Install stable helm repo

helm repo add stable https://kubernetes-charts.storage.googleapis.com

install the helm chart

helm install stable/spark-history-server --namespace default -f examples/spark-history-server-manifest.yaml --generate-name

spark-history-server-manifest.yaml

#s3 config
image:
  repository: spark-history-server
  tag: latest
  pullPolicy: Never
pvc:
  enablePVC: false
  existingClaimName: nfs-pvc
  eventsDir: “/”
nfs:
  enableExampleNFS: false
  pvName: nfs-pv
  pvcName: nfs-pvc
s3:
  enableS3: true
  enableIAM: false
  secret: aws-secrets
  accessKeyName: aws-access-key
  secretKeyName: aws-secret-key
  logDirectory: s3a://spark-events/logs/
  endpoint: http://minio-published:9000

The server should be accessible at http://localhost:18080/

Note: Have seen some issues with caching so will recommend incognito mode.

Step 5: Submit the spark job

kubectl apply -f examples/spark-pi.yaml

The event is minIO

The spark history server

In my earlier post, I package the application jar inside the docker image. For local development, it would be easier if we can run the code on K8 without rebuilding the image each time.

One way of doing that would be via sharing your local development path as a Volume. In this post, we will see how to do that

Package Spark Scala Code and Deploy it on Kubernetes using Spark-on-k8s-Operator

Versions:

1. Spark: 3.0.0
2. Scala: 2.12
3. SBT: 1.3.13
4. Docker On Mac: 2.2.0.0
5. Kubernetes: v1.15.5
6. spark-on-k8s-operator: sparkoperator.k8s.io/v1beta2

Step 1: Set up Kubernetes:

Please follow my earlier post to set up Kubernetes

Run Spark Job on Kubernetes using Spark-on-k8s-Operator

Step 2: The Project can be cloned from Github:

The project structure is as below:

Step 3: Set up for packaging:

1. The plugins.sbt file inside the project folder is required for building a flat jar.

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.10")

2. The build.properties file inside the project folder is required for providing the sbt version.

sbt.version=1.3.13

3. In build.sbt the task for merge strategy is required to resolve any conflicts during packaging

assemblyMergeStrategy in assembly := {
  case "META-INF/services/org.apache.spark.sql.sources.DataSourceRegister" => MergeStrategy.concat
  case PathList("META-INF", xs@_*) => MergeStrategy.discard
  case "application.conf" => MergeStrategy.concat
  case x => MergeStrategy.first
}

Step 4: Create yaml file to deploy/run the code on K8

The file is available in examples/spark-scala-k8-app.yaml

Important elements in the yaml file

1. spec.image: “gcr.io/spark-operator/spark:v3.0.0”. It’s a static image
2. spec.imagePullPolicy: Never . Since we are pulling image from local.
3. spec.mainClass: com.AppK8Demo . Provide the path to your job class. AppK8Demo is a sample Job class from my Github repo.
4. Update the spec.volumes as per your work directory.

apiVersion: "sparkoperator.k8s.io/v1beta2"
kind: SparkApplication
metadata:
  name: spark-scala-file-k8-app
  namespace: default
spec:
  type: Scala
  mode: cluster
  image: "gcr.io/spark-operator/spark:v3.0.0"
  imagePullPolicy: Never
  mainClass: com.AppK8Demo
  mainApplicationFile: "local:///opt/spark/work-dir/jar/spark-scala-k8-app-assembly-0.1.jar"
  sparkVersion: "3.0.0"
  restartPolicy:
    type: Never
  volumes:
    - name: "app-volume"
      hostPath:
        path: "/Users/sugupta/Desktop/codebase/personal/spark-scala-k8-app/target/scala-2.12"
        type: Directory
    - name: "file-volume"
      hostPath:
        path: "/Users/sugupta/Desktop/codebase/personal/spark-scala-k8-app/src/main/resources"
        type: Directory
  driver:
    cores: 1
    coreLimit: "1200m"
    memory: "512m"
    labels:
      version: 3.0.0
    serviceAccount: spark
    volumeMounts:
      - name: "app-volume"
        mountPath: "/opt/spark/work-dir/jar"
      - name: "file-volume"
        mountPath: "/opt/spark/work-dir/src/main/resources"
  executor:
    cores: 1
    instances: 1
    memory: "512m"
    labels:
      version: 3.0.0
    volumeMounts:
      - name: "app-volume"
        mountPath: "/opt/spark/work-dir/jar"
      - name: "file-volume"
        mountPath: "/opt/spark/work-dir/src/main/resources"

Run the job

kubectl apply -f examples/spark-scala-k8-app.yaml

Hoped this worked for you too

Spring Scheduler is an easy and efficient way of scheduling the tasks. However, one caveat is in case of multiple instances it fires the task on every instance. This behavior is not ideal for use cases where the expected behavior is at most once at the same time.

I have been using quartz for scheduling the jobs but recently came across Shedlock which makes scheduling much easier.

Spring-Scheduler triggers all the instances, Shedlock initiates a race condition and whichever instances are able to win takes control over the task and other instances silently exit the task.

How to implement Shedlock?

Step 1: Create a table that would be used by Shedlock to trigger the race condition.

CREATE TABLE IF NOT EXISTS shedlock(name VARCHAR(64) NOT NULL, locked_at TIMESTAMP NOT NULL, lock_until TIMESTAMP NOT NULL, locked_by VARCHAR(255) NOT NULL, PRIMARY KEY (name));

Step 2: Enable Schedlock and create SheldLock Lock Provider.

import javax.sql.DataSource;
import net.javacrumbs.shedlock.core.LockProvider;
import net.javacrumbs.shedlock.provider.jdbctemplate.JdbcTemplateLockProvider;
import net.javacrumbs.shedlock.spring.annotation.EnableSchedulerLock;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;

@Configuration
@EnableSchedulerLock(defaultLockAtMostFor = "30s")
@EnableScheduling
public class ShedLockConfig {

  @Bean
  public LockProvider lockProvider(DataSource dataSource) {
    return new JdbcTemplateLockProvider(dataSource, "shedlock");
  }
}

Step 3 Create a Spring Scheduler task with Shedlock.

import java.text.SimpleDateFormat;
import java.util.Date;
import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class ScheduledCurrentTimeTask {

  private static final Logger log = LoggerFactory.getLogger(ScheduledCurrentTimeTask.class);

  private static final SimpleDateFormat dateFormat = new SimpleDateFormat("HH:mm:ss");

  @Scheduled(fixedRate = 5000)
  @SchedulerLock(name = "currentTimeTask", lockAtLeastFor = "5S", lockAtMostFor = "10M")
  public void currentTimeTask() {
    log.info("The time is now {}", dateFormat.format(new Date()));
  }

}

The three steps and now Spring Scheduler is At most once at the same time in Spring Scheduler.

Sample Code

Download the sample code from Github.

Requirement

In each environment, we need to support two Database schemas. The schema should not be hardcoded and take the environment name as a suffix.

application.yaml

Our configuration:

spring:
  datasource:
    url: jdbc:postgresql://${database_name}/${service_database_name}
    username: ${database_user}
    password: ${database_password}
  jpa:
    properties:
      hibernate:
        dialect: org.hibernate.dialect.PostgreSQLDialect
        default_schema: test_${spring.profiles.active}
  flyway:
    schemas: test_${spring.profiles.active}
    out-of-order: true
    baseline-on-migrate: true
    placeholder-prefix: $$$
    placeholder-suffix: $$$
    placeholders:
        raw_schema: test_raw_${spring.profiles.active}

Default Flyway Schema: This is were flyway scripts will be executed.

flyway:
    schemas: ${database_schema_name}

Second Schema: This is the second schema which we will use in our flyway scripts.

The default prefix is ‘${‘, which is not suitable if you want to use spring properties. So I have updated the placeholder prefix and suffix as we need to use spring properties.

spring:
  flyway:
    placeholder-prefix: $$$
    placeholder-suffix: $$$
    placeholders:
        raw_schema: test_raw_${spring.profiles.active}

Flyway Script

Now you can use the placeholder like this:

create schema if not exists $$$raw_schema$$$;

Debezium

We would use the Debezium MySQL connector for reading MySQL Database logs. Debezium would put the events in Kafka.

Kafka-Connect

We would use the JDBC Connector to read the events from the Kafka Topic and put in PostgreSQL.

Code

The code can be downloaded from Github.

Build the pipeline

We would use docker-compose to set up the following:

1. MySQL: The source DB.
2. PostgreSQL: The destination DB
3. Kafka-Connect(Debezium and JDBC Connector): Debezium for reading MySQL Logs and JDBC Connector for pushing the change to PostgreSQL.
4. Kafka
5. Zookeeper

Build the image

To build the Kafka-connect, we will add the JDBC connector and PostgreSQL JDBC Driver to our Kafka-Connect image.

docker-compose build

Dockerfile

FROM debezium/connect:1.0
ENV KAFKA_CONNECT_JDBC_DIR=$KAFKA_CONNECT_PLUGINS_DIR/kafka-connect-jdbc

ARG POSTGRES_VERSION=42.2.8
ARG KAFKA_JDBC_VERSION=5.5.0

# Deploy PostgreSQL JDBC Driver
RUN cd /kafka/libs && curl -sO https://jdbc.postgresql.org/download/postgresql-$POSTGRES_VERSION.jar

# Deploy Kafka Connect JDBC
RUN mkdir $KAFKA_CONNECT_JDBC_DIR && cd $KAFKA_CONNECT_JDBC_DIR &&\
 curl -sO https://packages.confluent.io/maven/io/confluent/kafka-connect-jdbc/$KAFKA_JDBC_VERSION/kafka-connect-jdbc-$KAFKA_JDBC_VERSION.jar

Run the containers

docker-compose up -d

MySQL

MySQL comes with a database inventory.

Debezium Connector

Execute the following curl command to set up the Debezium connector for reading the logs from MySQL.

This will create topics following topics:

1. orders
2. addresses
3. customers
4. products
5. products_on_hand

curl -X POST \
  http://localhost:8083/connectors/ \
  -H 'Cache-Control: no-cache' \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "inventory-source",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "mysql",
        "database.port": "3306",
        "database.user": "debezium",
        "database.password": "dbz",
        "database.server.id": "184054",
        "database.server.name": "dbserver1",
        "database.whitelist": "inventory",
        "database.history.kafka.bootstrap.servers": "kafka:9092",
        "database.history.kafka.topic": "schema-changes.inventory",
        "decimal.handling.mode": "double",
        "snapshot.mode": "when_needed",
        "transforms": "route",                                                    
        "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter", 
        "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",                     
        "transforms.route.replacement": "$3"   
    }
}'

JDBC connector

Execute the following curl command to set up the JDBC connector for writing the events from “orders” MySQL Table to PostgreSQL.

We have kept “auto.create”: “true” so that it automatically creates tables in PostgreSQL.

curl -X POST \
  http://localhost:8083/connectors/ \
  -H 'Cache-Control: no-cache' \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "rtw-connector-sink",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "topics": "orders",
        "connection.url": "jdbc:postgresql://postgresql:5432/test_db",
        "connection.user": "test_user",
        "connection.password": "Welcome123",
        "transforms": "unwrap",                                                
        "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",   
        "auto.create": "true",                                                   
        "insert.mode": "insert",                                                 
        "pk.fields": "order_number",                                                     
        "pk.mode": "none"                  
     
    }
}'

PostgreSQL

Once the connector is set up it will automatically create a table with name orders.

Play Around

Make changes in MySQL orders table and see them getting reflected in PostgreSQL orders table.

Bird’s eye view of the architecture:

The focus of the post will be entirely on Integration Test

For Set Up And Running the Spark Application, please read my earlier post.

Spark Streaming with Kafka-Connect Debezium Connector

The Code

1. Spark: 2.4.5
2. Scala 2.12.8
3. Test Containers: 0.37.0
4. Scala-Test: 3.0.1

Project Structure

The Spark Application

The Debezium Connector reads MySQL DB changes from the Binlog file and pushes the changes as Debezium events to a Kafka Topic.

The Spark Application Reads the data from the Kafka Topic and pushes the change to PostgreSQL.

package com.sg.job.streaming

import com.sg.wrapper.SparkSessionWrapper
import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object StreamingJob extends App with SparkSessionWrapper {

  val currentDirectory = new java.io.File(".").getCanonicalPath
  val kafkaReaderConfig = KafkaReaderConfig("localhost:29092", "dbserver1.inventory.orders")
  val jdbcConfig = JDBCConfig(url = "jdbc:postgresql://localhost:5432/test")
  new StreamingJobExecutor(spark, kafkaReaderConfig, currentDirectory + "/checkpoint/job", jdbcConfig).execute()
}

case class JDBCConfig(url: String, user: String = "test", password: String = "Test123", tableName: String = "orders_it")

case class KafkaReaderConfig(kafkaBootstrapServers: String, topics: String, startingOffsets: String = "latest")

case class StreamingJobConfig(checkpointLocation: String, kafkaReaderConfig: KafkaReaderConfig)

class StreamingJobExecutor(spark: SparkSession, kafkaReaderConfig: KafkaReaderConfig, checkpointLocation: String, jdbcConfig: JDBCConfig) {

  def execute(): Unit = {
    // read data from kafka and parse them
    val transformDF = read()
      .selectExpr("CAST(key AS STRING) as key", "CAST(value AS STRING) as value", "topic")

    transformDF
      .writeStream
      .option("checkpointLocation", checkpointLocation)
      .foreachBatch { (batchDF: DataFrame, _: Long) => {
        batchDF.write
          .format("jdbc")
          .option("url", jdbcConfig.url)
          .option("user", jdbcConfig.user)
          .option("password", jdbcConfig.password)
          .option("driver", "org.postgresql.Driver")
          .option(JDBCOptions.JDBC_TABLE_NAME, jdbcConfig.tableName)
          .option("stringtype", "unspecified")
          .mode(SaveMode.Append)
          .save()
      }
      }.start()
      .awaitTermination()
  }

  def read(): DataFrame = {
    spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", kafkaReaderConfig.kafkaBootstrapServers)
      .option("subscribe", kafkaReaderConfig.topics)
      .option("startingOffsets", kafkaReaderConfig.startingOffsets)
      .load()
  }
}

Integration Test For the Spark Application

For writing the integration test for the Spark Application discussed above we need the following:

1. MySQL DB: With Binary Logging
2. Kafka-Connect with Debezium connector
3. Kafka
4. Zookeeper
5. PostgreSQL

I will use Docker Test Containers to build the required infra.

build.sbt

name := "spark-streaming-with-debezium"

version := "0.1"
scalaVersion := "2.12.8"

val sparkVersion = "2.4.5"
val testcontainersScalaVersion = "0.37.0"

resolvers += "confluent" at "http://packages.confluent.io/maven/"
resolvers += Resolver.jcenterRepo
resolvers += "Spark Packages Repo" at "http://dl.bintray.com/spark-packages/maven"


assemblyMergeStrategy in assembly := {
  case "META-INF/services/org.apache.spark.sql.sources.DataSourceRegister" => MergeStrategy.concat
  case PathList("META-INF", xs@_*) => MergeStrategy.discard
  case "application.conf" => MergeStrategy.concat
  case x => MergeStrategy.first
}

coverageEnabled.in(ThisBuild, IntegrationTest, test) := true

//skipping test cases during package
test in assembly := {}

lazy val server = (project in file("."))
  .configs(IntegrationTest)
  .settings(Defaults.itSettings)

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-sql" % sparkVersion,
  "org.apache.spark" %% "spark-sql-kafka-0-10" % sparkVersion,
  "org.apache.kafka" % "kafka-clients" % "2.2.1",
  "org.postgresql" % "postgresql" % "42.2.5",
  "com.typesafe" % "config" % "1.4.0",
  "org.slf4j" % "slf4j-simple" % "1.7.30",
  "org.scalatest" % "scalatest_2.12" % "3.0.1" % "it,test",
  "com.dimafeng" %% "testcontainers-scala-scalatest" % testcontainersScalaVersion % IntegrationTest,
  "com.dimafeng" %% "testcontainers-scala-postgresql" % testcontainersScalaVersion % IntegrationTest,
  "com.dimafeng" %% "testcontainers-scala-kafka" % testcontainersScalaVersion % IntegrationTest,
  "com.dimafeng" %% "testcontainers-scala-mysql" % testcontainersScalaVersion % IntegrationTest,
  "io.debezium" % "debezium-testing-testcontainers" % "1.0.3.Final" % IntegrationTest,
  "mysql" % "mysql-connector-java" % "5.1.49" % IntegrationTest
)

Integration Test Flow

1. At the start of the test case execution, we will start our docker test containers.
2. We will do some write operations on our MySQL Database.
3. Will add Debezium connector via a Curl command
4. We will wait for some time to ensure that Kafka Connector is ready to read messages from MySQL DB
5. Kafka-Connect(Debezium) Connector will read the MySQL changes and put them as events in Kafka.
6. We will execute our Spark Structured Streaming job.
7. The job will stream the Kafka messages and with small transformation put them in PostgreSQL.
8. After some time a thread will stop the streaming job.
9. Now, we will assert the records in PostgreSQL with the expected data.

Integration Test class

package com.sg.job.streaming

import java.sql.{Connection, DriverManager, Statement}
import java.time.LocalDateTime

import com.dimafeng.testcontainers.{ForAllTestContainer, MultipleContainers}
import com.sg.wrapper.{ContainerTestWrapper, SparkSessionITWrapper}
import io.debezium.testing.testcontainers.ConnectorConfiguration
import org.apache.spark.sql.SparkSession
import org.scalatest.{BeforeAndAfterAll, FlatSpec}
import org.slf4j.{Logger, LoggerFactory}

/**
 * A Integration Test for [[StreamingJob]]
 */
class StreamingIT extends FlatSpec with ForAllTestContainer with BeforeAndAfterAll with ContainerTestWrapper with SparkSessionITWrapper {

  val sparkTest: SparkSession = spark
  override val container: MultipleContainers = MultipleContainers(kafkaContainer, postgresContainer, mySQLContainer)
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName)

  it must "integration test for streaming job" in {

    logger.info("Kafka-Connect is ready to send events...")
    createBinLogDBRecords(mySQLContainer.driverClassName, mySQLContainer.jdbcUrl, mySQLContainer.username, mySQLContainer.password)

    logger.info("launching the job...")

    val checkpoint = new java.io.File(".").getCanonicalPath + "/checkpoint/it" + LocalDateTime.now()
    val kafkaReaderConfig = KafkaReaderConfig("localhost:29092", "dbserver1.inventory.orders")
    val jobcConfig = JDBCConfig(url = postgresContainer.jdbcUrl)

    new StreamingJobExecutor(sparkTest, kafkaReaderConfig, checkpoint, jobcConfig).execute()

    logger.info("Assert the records")
    Class.forName(postgresContainer.driverClassName)
    val connection: Connection = DriverManager.getConnection(postgresContainer.jdbcUrl, postgresContainer.username, postgresContainer.password)
    connection.setAutoCommit(true)
    connection.setSchema("test")
    val statement: Statement = connection.createStatement()

    val sql = "select * from public.orders_it"
    val rs = statement.executeQuery(sql)
    assert(rs != null)
    try {
      while (rs.next()) {
        //Put the logic for assertion
        println(rs.getString("key"), rs.getString("value"), rs.getString("topic"))
      }
    } finally {
      rs.close()
    }
  }

  override def beforeAll() {
    super.beforeAll()
    debeziumContainer.start()
    createSnapshotDBRecords(mySQLContainer.driverClassName, mySQLContainer.jdbcUrl, mySQLContainer.username, mySQLContainer.password)

    val connector = ConnectorConfiguration.forJdbcContainer(mySQLContainer.container)
      .`with`("database.server.name", "dbserver1")
      .`with`("connector.class", "io.debezium.connector.mysql.MySqlConnector")
      .`with`("database.whitelist", "inventory")
      .`with`("database.user", "root")
      .`with`("database.password", "test")
      .`with`("database.hostname", mySQLContainer.networkAliases.head)
      .`with`("database.server.id", "123")
      .`with`("database.port", "3306")
      .`with`("name", "my-connector")
      .`with`("database.history.kafka.bootstrap.servers", s"${kafkaContainer.networkAliases.head}:9092")
      .`with`("database.history.kafka.topic", "schema-changes.inventory")

    logger.info("Trying to connect to Kafka-Connect")
    debeziumContainer.registerConnector("my-connector", connector)

    //Giving time for kafka-connect to settle down.
    Thread.sleep(30000)

    //A way out of stopping the streaming job
    val runnable: Runnable = () =>
      try {
        Thread.sleep(60000)
        logger.info("Stopping the job...")
        sparkTest.streams.active.foreach(q => q.stop())
        sparkTest.sparkContext.stop()
        sparkTest.stop()
      } catch {
        case e: InterruptedException => e.printStackTrace()
      }
    val thread = new Thread(runnable)
    thread.start()
  }

  override def afterAll() {
    sparkTest.stop()
    debeziumContainer.stop()
    super.afterAll()
  }

  /**
   * DB changes for snapshot mode
   *
   */
  def createSnapshotDBRecords(driverClassName: String, jdbcUrl: String, username: String, password: String): Unit = {
    Class.forName(driverClassName)
    val connection: Connection = DriverManager.getConnection(jdbcUrl, username, password)
    val statement: Statement = connection.createStatement()
    statement.execute("create table inventory.orders (id int not null, title varchar(255), primary key (id))")
    statement.execute("insert into inventory.orders values (1, 'Learn ADC')")
    statement.execute("insert into inventory.orders values (2, 'Learn BDC')")
    statement.closeOnCompletion()
    connection.close()

  }
  /**
   * DB change for binlog mode
   *
   */
  def createBinLogDBRecords(driverClassName: String, jdbcUrl: String, username: String, password: String): Unit = {
    Class.forName(driverClassName)
    val connection: Connection = DriverManager.getConnection(jdbcUrl, username, password)
    val statement: Statement = connection.createStatement()
    statement.execute("insert into inventory.orders values (3, 'Learn CDC')")
    statement.execute("insert into inventory.orders values (4, 'Learn DDC')")
    statement.execute("update inventory.orders set title = 'Learn EDC' where id = 4")
    statement.execute("delete from inventory.orders where id = 3")
    statement.closeOnCompletion()
    connection.close()
  }

}

When you run the test case, it will first bring the containers

Once the containers are up, we insert or test data in MySQL.

Debezium Connector ensures that MySQL records are available in Kafka as events.

Spark Application reads the Kafka topic and after doing the required transformation inserts data to PostgreSQL.

After 90 seconds, spark application is killed and control comebacks to Integration Test.

Now, you can create a connection to PostgreSQL DB and perform the required assertion.

sbt

Execute the following command to run the integration test

sbt clean coverage test coverage it:test coverageReport

I hope this was precise and helpful. The code can be downloaded from Github

Bird’s eye view of the architecture:

Debezium

Debezium is an open-source distributed platform for change data capture. It provides a set of Kafka connector which are capable of reading from the Database Binlog files and produce the changes as events in Kafka.

As a CDC, it gives you a state of Database Row before & after the commit and also tells what kind of operation it was i.e Create/ Update or Delete.

A Debezium Kafka message looks like:

{
   "schema":{
      "type":"struct",
      "fields":[
         {
            "type":"struct",
            "fields":[
               {
                  "type":"int32",
                  "optional":false,
                  "field":"order_number"
               },
               {
                  "type":"int32",
                  "optional":false,
                  "name":"io.debezium.time.Date",
                  "version":1,
                  "field":"order_date"
               },
               {
                  "type":"int32",
                  "optional":false,
                  "field":"purchaser"
               },
               {
                  "type":"int32",
                  "optional":false,
                  "field":"quantity"
               },
               {
                  "type":"int32",
                  "optional":false,
                  "field":"product_id"
               }
            ],
            "optional":true,
            "name":"dbserver1.inventory.orders.Value",
            "field":"before"
         },
         {
            "type":"struct",
            "fields":[
               {
                  "type":"int32",
                  "optional":false,
                  "field":"order_number"
               },
               {
                  "type":"int32",
                  "optional":false,
                  "name":"io.debezium.time.Date",
                  "version":1,
                  "field":"order_date"
               },
               {
                  "type":"int32",
                  "optional":false,
                  "field":"purchaser"
               },
               {
                  "type":"int32",
                  "optional":false,
                  "field":"quantity"
               },
               {
                  "type":"int32",
                  "optional":false,
                  "field":"product_id"
               }
            ],
            "optional":true,
            "name":"dbserver1.inventory.orders.Value",
            "field":"after"
         },
         {
            "type":"struct",
            "fields":[
               {
                  "type":"string",
                  "optional":false,
                  "field":"version"
               },
               {
                  "type":"string",
                  "optional":false,
                  "field":"connector"
               },
               {
                  "type":"string",
                  "optional":false,
                  "field":"name"
               },
               {
                  "type":"int64",
                  "optional":false,
                  "field":"ts_ms"
               },
               {
                  "type":"string",
                  "optional":true,
                  "name":"io.debezium.data.Enum",
                  "version":1,
                  "parameters":{
                     "allowed":"true,last,false"
                  },
                  "default":"false",
                  "field":"snapshot"
               },
               {
                  "type":"string",
                  "optional":false,
                  "field":"db"
               },
               {
                  "type":"string",
                  "optional":true,
                  "field":"table"
               },
               {
                  "type":"int64",
                  "optional":false,
                  "field":"server_id"
               },
               {
                  "type":"string",
                  "optional":true,
                  "field":"gtid"
               },
               {
                  "type":"string",
                  "optional":false,
                  "field":"file"
               },
               {
                  "type":"int64",
                  "optional":false,
                  "field":"pos"
               },
               {
                  "type":"int32",
                  "optional":false,
                  "field":"row"
               },
               {
                  "type":"int64",
                  "optional":true,
                  "field":"thread"
               },
               {
                  "type":"string",
                  "optional":true,
                  "field":"query"
               }
            ],
            "optional":false,
            "name":"io.debezium.connector.mysql.Source",
            "field":"source"
         },
         {
            "type":"string",
            "optional":false,
            "field":"op"
         },
         {
            "type":"int64",
            "optional":true,
            "field":"ts_ms"
         }
      ],
      "optional":false,
      "name":"dbserver1.inventory.orders.Envelope"
   },
   "payload":{
      "before":{
         "order_number":10004,
         "order_date":16852,
         "purchaser":1003,
         "quantity":1,
         "product_id":107
      },
      "after":{
         "order_number":10004,
         "order_date":16852,
         "purchaser":1003,
         "quantity":3,
         "product_id":107
      },
      "source":{
         "version":"1.0.3.Final",
         "connector":"mysql",
         "name":"dbserver1",
         "ts_ms":1589660547000,
         "snapshot":"false",
         "db":"inventory",
         "table":"orders",
         "server_id":223344,
         "gtid":null,
         "file":"mysql-bin.000003",
         "pos":354,
         "row":0,
         "thread":12,
         "query":null
      },
      "op":"u",
      "ts_ms":1589660547884
   }
}

Spark Streaming

We are using Spark-Streaming as a processing engine, it reads the Debezium events from the Kafka topic and pushes the changes to PostgreSQL.

Sample Code

The sample code under discussion can be cloned from Github

How to bring the infrastructure up and running

docker-compose

version: '3.0'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    container_name: zookeeper
    ports:
      - 2181:2181
      - 2888:2888
      - 3888:3888
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
      - ZOOKEEPER_CLIENT_PORT=2181
      - ZOOKEEPER_TICK_TIME=2000
  kafka:
    image: confluentinc/cp-kafka:latest
    container_name: kafka
    ports:
      - 9092:9092
      - 29092:29092
    depends_on:
      - zookeeper
    environment:
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      - KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
  mysql:
    image: debezium/example-mysql:1.0
    container_name: mysql
    ports:
      - 3306:3306
    environment:
      - MYSQL_ROOT_PASSWORD=debezium
      - MYSQL_USER=mysqluser
      - MYSQL_PASSWORD=mysqlpw
  connect:
    image: debezium/connect:1.0
    container_name: connect
    ports:
      - 8083:8083
    depends_on:
      - kafka
      - mysql
    environment:
      - BOOTSTRAP_SERVERS=kafka:9092
      - GROUP_ID=1
      - CONFIG_STORAGE_TOPIC=my_connect_configs
      - OFFSET_STORAGE_TOPIC=my_connect_offsets
      - STATUS_STORAGE_TOPIC=my_connect_statuses
  postgresql:
    image: sameersbn/postgresql:9.4
    container_name: postgresql
    environment:
      - DEBUG=false
      - DB_USER=test
      - DB_PASS=Test123
      - DB_NAME=test
    ports:
      - "5432:5432"

Execute:

docker-compose up -d

It will bring up the following containers:

Add Debezium-Connector

Once the container “connect” is up and running, execute the following command to add MySqlConnector:

curl -X POST \
  http://localhost:8083/connectors/ \
  -H 'Cache-Control: no-cache' \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "inventory-connector",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "mysql",
        "database.port": "3306",
        "database.user": "debezium",
        "database.password": "dbz",
        "database.server.id": "1",
        "database.server.name": "dbserver1",
        "database.whitelist": "inventory",
        "database.history.kafka.bootstrap.servers": "kafka:9092",
        "database.history.kafka.topic": "schema-changes.inventory",
        "decimal.handling.mode": "double",
        "snapshot.mode": "when_needed",
        "table.whitelist":"inventory.orders"
    }
}'

The connector is now ready to read MySQL Binlog files.

MySQL Database

The MySQL database in the docker-compose comes with a sample database and has the required database changes which are required for the connector to read the DB logs.

If you plan to use your own database then ensure the following things:

1. The MySQL config for ‘binlog_format’ should be ROW

show variable 'binlog_format'; //should be ROW

2. Binary logging should be enabled. In MySQL cnf file:

# Uncomment the following if you want to log updates
log-bin=mysql-bin
expire_logs_days  = 10
binlog_format     = row

Kafka Topic

Once the connector is added via curl command, it starts pushing the snapshot changes to Kafka, and once that is done it changes it’s mode to Binlog where it is watching the changes in Binlog file and pushed them to Kafka.

Kafka Topic:

kafkacat -b localhost:29092 -t dbserver1.inventory.orders

Spark Streaming Job

The Spark-Streaming will listen to the Kafka Message, serialize/deserialize the message and push the change to PostgreSQL.

package com.sg.job.streaming

import com.sg.wrapper.SparkSessionWrapper
import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object StreamingJob extends App with SparkSessionWrapper {

  val currentDirectory = new java.io.File(".").getCanonicalPath
  val kafkaReaderConfig = KafkaReaderConfig("localhost:29092", "dbserver1.inventory.orders")
  val jdbCConfig = JDBCConfig(url = "jdbc:postgresql://localhost:5432/test")
  new StreamingJobExecutor(spark, kafkaReaderConfig, currentDirectory + "/checkpoint/job", jdbCConfig).execute()
}

case class JDBCConfig(url: String, user: String = "test", password: String = "Test123", tableName: String = "orders_it")

case class KafkaReaderConfig(kafkaBootstrapServers: String, topics: String, startingOffsets: String = "latest")

case class StreamingJobConfig(checkpointLocation: String, kafkaReaderConfig: KafkaReaderConfig)

class StreamingJobExecutor(spark: SparkSession, kafkaReaderConfig: KafkaReaderConfig, checkpointLocation: String, jdbcConfig: JDBCConfig) {

  def execute(): Unit = {
    // read data from kafka and parse them
    val transformDF = read()
      .selectExpr("CAST(key AS STRING) as key", "CAST(value AS STRING) as value", "topic")

    transformDF
      .writeStream
      .option("checkpointLocation", checkpointLocation)
      .foreachBatch { (batchDF: DataFrame, _: Long) => {
        batchDF.write
          .format("jdbc")
          .option("url", jdbcConfig.url)
          .option("user", jdbcConfig.user)
          .option("password", jdbcConfig.password)
          .option("driver", "org.postgresql.Driver")
          .option(JDBCOptions.JDBC_TABLE_NAME, jdbcConfig.tableName)
          .option("stringtype", "unspecified")
          .mode(SaveMode.Append)
          .save()
      }
      }.start()
      .awaitTermination()
  }

  def read(): DataFrame = {
    spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", kafkaReaderConfig.kafkaBootstrapServers)
      .option("subscribe", kafkaReaderConfig.topics)
      .option("startingOffsets", kafkaReaderConfig.startingOffsets)
      .load()
  }
}

PostgreSQL

Once the job is running, you can connect to PostgreSQL DB and see a table with the name orders_it has been created.

Now, play around with MySQL DB and see the changes getting pushed to PostgreSQL.

The same can be simulated via an Integration Test.

Integration Testing in Spark Structured Streaming