Run OLake Sync on Kubernetes Using Apache Airflow | Guide

Running OLake on Kubernetes with Your Existing Airflow, Sync Your Data Effortlessly

At OLake Go, we're building tools to make data integration seamless. Today, we're excited to show you how to leverage your existing Apache Airflow setup to automate OLake Go data synchronization tasks directly on your Kubernetes cluster!

OLake Go is designed to efficiently sync data from various sources to your chosen destinations. This guide provides an Airflow DAG (Directed Acyclic Graph) that orchestrates the OLake Go sync command, handling dependencies like persistent storage and configuration management within Kubernetes.

This post assumes you already have:

1. A running Apache Airflow instance.

Airflow Kubernetes Provider: The necessary provider must be installed in your Airflow environment. If not already installed, run:

bash

# Activate your Airflow virtual environment first
pip install apache-airflow-providers-cncf-kubernetes

# Restart Airflow scheduler and webserver after installation

2. Access to a Kubernetes cluster.

3. Basic familiarity with Airflow concepts (UI, DAGs, Connections) and Kubernetes concepts (Namespaces, ConfigMaps, StorageClasses, PersistentVolumeClaims).

Let's dive in!

Setting the Stage

Before deploying the DAG, ensure the following are in place:

Airflow with LocalExecutor:

info

Crucially, your Airflow instance must be configured to use the LocalExecutor or SequentialExecutor.

Our DAG utilizes Airflow's PythonOperator to dynamically check for and create Kubernetes resources (like PersistentVolumeClaims) before running the main OLake Go task.

This Python code runs directly on the machine where the Airflow scheduler is running, which is how LocalExecutor and SequentialExecutor operate.

It will not work correctly out-of-the-box with the CeleryExecutor or KubernetesExecutor without modification.

You can verify this by checking the executor setting under the [core] section in your $AIRFLOW_HOME/airflow/airflow.cfg file. It should be set to either executor = LocalExecutor or executor = SequentialExecutor.

Kubernetes Connection in Airflow

You need an Airflow Connection configured to communicate with your Kubernetes cluster.

1. Go to Admin -> Connections in the Airflow UI.

2. Add or identify a connection of type Kubernetes.

3. Configure it using either a Kubeconfig file (pasted JSON recommended for setups outside the cluster). To get the kubeconfig on your machine in JSON format. Please run:

bash

kubectl config view --raw --flatten --output='json'

tip

Note down the Connection Id - you'll need it for the DAG configuration.

Kubernetes Namespace

Choose or create a namespace in your Kubernetes cluster where OLake pods and related resources (like the PVC) will reside. You'll need the name of this namespace.

Kubernetes ConfigMaps

The OLake Go DAG relies on Kubernetes ConfigMaps to inject configuration. You must create three ConfigMaps in the target namespace before running the DAG:

1. olake-source-config: Containing your source configuration in a key named source.json.

   Download cm_olake-source-config.yaml

bash

curl -Lo cm_olake-source-config.yaml https://raw.githubusercontent.com/datazip-inc/olake-docs/refs/heads/master/kubernetes/cm_olake-source-config.yaml

2. olake-destination-config: Containing your destination configuration in a key named destination.json.

   Download cm_olake-destination-config.yaml

bash

curl -Lo cm_olake-destination-config.yaml https://raw.githubusercontent.com/datazip-inc/olake-docs/refs/heads/master/kubernetes/cm_olake-destination-config.yaml

3. olake-streams-config: Containing your OLake streams configuration in a key named streams.json

   Download cm_olake-streams-config.yaml

bash

curl -Lo cm_olake-streams-config.yaml https://raw.githubusercontent.com/datazip-inc/olake-docs/refs/heads/master/kubernetes/cm_olake-streams-config.yaml

info

This setup requires a streams.json generated beforehand using the OLake Go discover command against your source database.

• Streams Generation Guides:
  • Streams config
• The content of this file will be placed within the cm_olake-streams-config.yaml ConfigMap.

Kubernetes StorageClass

OLake Go uses a state file to keep track of sync progress. To persist this state between runs, the DAG creates a PersistentVolumeClaim (PVC). You need a StorageClass in your Kubernetes cluster that supports ReadWriteMany (RWX) access mode.

note

At start of the sync, the state file just contains {}. OLake Go will update this file as it processes data.

Examples: azurefile on AKS, Google Cloud Filestore provisioner on GKE, AWS EFS provisioner on EKS, or an NFS provisioner.

Identify the name of a suitable StorageClass. You can often list available StorageClasses using kubectl get storageclass. Consult your Kubernetes administrator if unsure.

Getting the DAG File

You'll need the Python DAG file that defines the workflow. Download it here.

Configuring the DAG

Now, open the downloaded olake_sync_from_source.py file in a text editor. You need to customize a few variables at the top of the file to match your specific environment. Do not skip this step!

python

# --- Configuration ---
# !!! IMPORTANT: Set this to the ID of your configured Kubernetes connection in Airflow !!!
# This connection tells Airflow how to authenticate with your K8s cluster.
KUBERNETES_CONN_ID = "kubernetes_default" # <-- EDIT THIS LINE

# !!! IMPORTANT: Set this to the Kubernetes namespace where OLake pods should run !!!
# Ensure ConfigMaps and the PVC exist or will be created in this namespace.
TARGET_NAMESPACE = "olake" # <-- EDIT THIS LINE

# !!! IMPORTANT: Set this to the correct OLake image for your source database !!!
# Find images at: https://hub.docker.com/u/olakego
# Examples: "olakego/source-mongodb:latest", "olakego/source-mysql:latest", "olakego/source-postgres:latest"
OLAKE_IMAGE = "olakego/source-db:latest" # <-- EDIT THIS LINE

# !!! IMPORTANT: Set this to the name of a StorageClass in your K8s cluster !!!
# This StorageClass MUST support ReadWriteMany (RWX) access mode.
# Examples: "azurefile" (AKS), "standard-rwx" (GKE Filestore), an EFS provisioner (EKS)
STORAGE_CLASS_NAME = "default" # <-- EDIT THIS LINE

# --- Optional Node Affinity ---
# Set NODE_AFFINITY_LABEL_KEY and NODE_AFFINITY_LABEL_VALUE to target specific nodes.
# Set NODE_AFFINITY_REQUIRED to True for hard requirement, False for preference.
NODE_AFFINITY_REQUIRED = False # <-- EDIT THIS LINE (True/False)
NODE_AFFINITY_LABEL_KEY = "nodegroup_name" # <-- EDIT THIS LINE (e.g., "topology.kubernetes.io/zone")
NODE_AFFINITY_LABEL_VALUE = "olake" # <-- EDIT THIS LINE (e.g., "us-central1-a")

# --- Names of ConfigMaps ---
# Ensure ConfigMaps with these names exist in the TARGET_NAMESPACE
# containing your source, destination, and streams JSON configurations respectively.
SOURCE_CONFIG_MAP_NAME = "olake-source-config"
DESTINATION_CONFIG_MAP_NAME = "olake-destination-config"
STREAMS_CONFIG_MAP_NAME = "olake-streams-config"

Recap of values to change:

• KUBERNETES_CONN_ID: Set this to the exact "Connection Id" of your Kubernetes connection in Airflow.

• TARGET_NAMESPACE: Set this to the Kubernetes namespace where you created the ConfigMaps and where the OLake pod/PVC should run.

• OLAKE_IMAGE: Crucially, change this to the specific OLake Docker image corresponding to your data source (e.g., olakego/source-postgres:latest, olakego/source-mongodb:latest, etc). Check the OLake Docker Hub page for available images and tags.

• STORAGE_CLASS_NAME: Set this to the name of an available ReadWriteMany (RWX) StorageClass in your cluster.

• NODE_AFFINITY_* (Optional): Adjust these if you need to control pod placement onto specific Kubernetes nodes based on labels. Leave NODE_AFFINITY_REQUIRED = False if you don't need strict node requirements.

• *_CONFIG_MAP_NAME (Optional): Only change these if you named your prerequisite ConfigMaps differently than the defaults.

Deploying the DAG to Airflow

1. Save your modified olake_sync_from_source.py file.

2. Place the file into the dags folder recognized by your Airflow instance. The location of this folder depends on your Airflow setup.

3. Airflow automatically scans this folder. Wait a minute or two, and the DAG named olake_sync_from_source should appear in the Airflow UI. You might need to unpause it (toggle button on the left) if it loads in a paused state.

Running Your OLake Go Sync

1. Navigate to the Airflow UI.

2. Find the olake_sync_from_source DAG.

3. Click the "Play" button (▶️) on the right side to trigger a manual run.

4. Click on the running DAG instance to view the Graph or Gantt chart.

5. The first task (create_pvc) will run, checking for and creating the olake-shared-data PVC if needed.

6. Once that succeeds, the sync_data task will start. This task launches the actual OLake pod on your Kubernetes cluster.

7. You can click on the sync_data task instance and view its logs to see the output from the OLake Go process itself.

How to set up OLake Go cron job scheduler

Modify the line #52 from the olake_sync_from_source.py DAG file (schedule=None) with the frequency you wish to setup.

python

with DAG(
    dag_id="olake_sync_from_source",
    start_date=dag_start_date,
    schedule="@daily",  # SAMPLE VALUE
    catchup=False,
    ...)

For more information on how to add a schedule, refer Cron & Time Intervals docs.

That's it! You've successfully configured and run an OLake Go sync task orchestrated by your existing Airflow instance, leveraging the power of Kubernetes for execution. This setup provides a robust and automated way to manage your data synchronization pipelines.

Happy Syncing!

FAQs

Q1. How do I automate OLake Go data sync jobs with Apache Airflow on Kubernetes?

Create an Airflow DAG that uses the KubernetesPodOperator from the apache-airflow-providers-cncf-kubernetes package. The DAG defines the OLake Go sync task as a Kubernetes pod that:

• Mounts OLake Go configuration from a ConfigMap
• Uses a PersistentVolumeClaim for sync state storage across runs

Airflow's scheduler triggers the DAG on your defined schedule, dynamically creating OLake pods in the cluster for each sync run and cleaning them up on completion.

Q2. What Airflow executor is required to run OLake Go's Kubernetes integration?

This DAG requires the LocalExecutor or SequentialExecutor.

The reason is the DAG's first task, create_pvc, which uses Airflow's PythonOperator to check for and create Kubernetes resources (such as the PersistentVolumeClaim) before the sync runs. That Python code executes directly on the machine running the Airflow scheduler, which is how the LocalExecutor and SequentialExecutor operate. It will not work correctly out of the box with the CeleryExecutor or KubernetesExecutor without modification.

You can verify your setting under the [core] section of $AIRFLOW_HOME/airflow/airflow.cfg: it should read executor = LocalExecutor or executor = SequentialExecutor.

Note that this constraint comes from the DAG's PythonOperator step, not from the KubernetesPodOperator itself, which launches pods via the Kubernetes API regardless of executor.

Q3. How does OLake Go manage state between Airflow-triggered sync runs on Kubernetes?

OLake Go stores sync state - checkpoints, cursor positions, and schema snapshots - in a PersistentVolumeClaim mounted to each worker pod. The PVC is created once and reused across all sync runs.

The Airflow DAG includes a pre-task that checks whether the PVC already exists and creates it if not - this makes the setup idempotent and safe to re-run. If a sync fails mid-way, the next Airflow-triggered run resumes from the last checkpoint rather than starting a full reload.

Q4. What Kubernetes resources does the OLake Go Airflow DAG create and manage?

The DAG manages three types of Kubernetes resources:

1. ConfigMap - Contains OLake Go's source and destination configuration files, mounted into the pod as a volume
2. PersistentVolumeClaim - Provides durable state storage for sync progress, created once and reused across runs
3. OLake worker pods - Dynamically created by KubernetesPodOperator for each sync execution. The operator passes config via volume mounts, runs the OLake Go sync command, and cleans up the pod on completion (is_delete_operator_pod=True), keeping the cluster tidy between runs

Q5. What are the prerequisites for running OLake Go with Airflow on Kubernetes?

You need the following before setting up OLake Go with Airflow on Kubernetes:

• Apache Airflow with the Kubernetes provider installed:

pip install apache-airflow-providers-cncf-kubernetes

• Kubernetes cluster access with kubectl configured and pointing to your target cluster
• Airflow Kubernetes Connection - Create a connection of type Kubernetes Cluster Connection in Airflow's connection management UI, providing your kubeconfig in JSON format with the connection ID k8s_conn
• StorageClass that supports PersistentVolumeClaims - Required for OLake Go state persistence between sync runs
• OLake Go configuration files for your source database and destination (Iceberg catalog, S3 credentials, etc.)

OLake Go

Replicate databases, Kafka, and S3 into Apache Iceberg with OLake Go, an open source EL engine built for Iceberg from the ground up.

SignupJoin SlackExplore OLake GitHub

Contact us at hello@olake.io