I/O: Migrate Kafka->CrateDB recipe from ingestr to omniload

.github/dependabot.yml

@@ -173,6 +173,16 @@ updates:
     schedule:
       interval: "daily"
 
+  - directory: "/application/omniload"
+    package-ecosystem: "pip"
+    schedule:
+      interval: "daily"
+
+  - directory: "/application/omniload"
+    package-ecosystem: "docker-compose"
+    schedule:
+      interval: "daily"
+
   - directory: "/application/pgpool"
     package-ecosystem: "docker-compose"
     schedule:

.github/workflows/application-ingestr.yml

@@ -1,4 +1,4 @@
-name: "Ingestr"
+name: "ingestr"
 
 on:
   pull_request:
@@ -40,7 +40,11 @@ jobs:
         uses: actions/checkout@v6
 
       - name: Install uv
-        uses: astral-sh/setup-uv@v7
+        uses: astral-sh/setup-uv@v8.2.0
+        with:
+          activate-environment: 'true'
+          enable-cache: true
+          python-version: '3.14'
 
       - name: Validate application/ingestr
         run: |

.github/workflows/application-omniload.yml

@@ -0,0 +1,53 @@
+name: "omniload"
+
+on:
+  pull_request:
+    paths:
+    - '.github/workflows/application-omniload.yml'
+    - 'application/omniload/**'
+  push:
+    branches: [ main ]
+    paths:
+    - '.github/workflows/application-omniload.yml'
+    - 'application/omniload/**'
+
+  # Allow job to be triggered manually.
+  workflow_dispatch:
+
+  # Run job each night after CrateDB nightly has been published.
+  schedule:
+    - cron: '0 3 * * *'
+
+# Cancel in-progress jobs when pushing to the same branch.
+concurrency:
+  cancel-in-progress: true
+  group: ${{ github.workflow }}-${{ github.ref }}
+
+jobs:
+
+  test:
+    runs-on: ${{ matrix.os }}
+
+    strategy:
+      fail-fast: true
+      matrix:
+        os: [ "ubuntu-latest" ]
+
+    name: OS ${{ matrix.os }}
+    steps:
+
+      - name: Acquire sources
+        uses: actions/checkout@v6
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v8.2.0
+        with:
+          activate-environment: 'true'
+          enable-cache: true
+          python-version: '3.14'
+
+      - name: Validate application/omniload
+        run: |
+          cd application/omniload
+          uv pip install -r requirements.txt
+          sh test.sh

application/ingestr/.env

@@ -1,8 +1,3 @@
-# Data source configuration (Kafka).
-export PORT_KAFKA_BROKER_EXTERNAL=9092
-export PORT_KAFKA_BROKER_INTERNAL=9094
-export PORT_KAFKA_ZOOKEEPER=2181
-
 # Data sink configuration (CrateDB).
 export CRATEDB_HOST=${CRATEDB_HOST:-cratedb}
 export CRATEDB_HTTP_PORT=${CRATEDB_HTTP_PORT:-4200}

application/ingestr/README.md

@@ -12,8 +12,8 @@ They are also used as integration tests to ensure software components
 fit together well.
 
 - `.env`: Environment variable definitions
-- `kafka-compose.yml`: Service composition file for Kafka and CrateDB
-- `kafka-demo.xsh`: The demonstration program
+- `elasticsearch-compose.yml`: Service composition file for Elasticsearch and CrateDB
+- `elasticsearch-demo.sh`: The demonstration program
 
 ## Prerequisites
 
@@ -22,12 +22,22 @@ installation of Docker and Python.
 
 ## Synopsis
 
+Import data from CSV into Elasticsearch.
 ```shell
 ingestr ingest --yes \
-  --source-uri "kafka://?bootstrap_servers=localhost:9092&group_id=test_group&value_type=json&select=value" \
-  --source-table "demo" \
-  --dest-uri "cratedb://crate:crate@localhost:5432/?sslmode=disable" \
-  --dest-table "doc.kafka_demo"
+  --source-uri "csv://taxi_details.csv" \
+  --source-table "data" \
+  --dest-uri "elasticsearch://elasticsearch:9200?secure=false" \
+  --dest-table "example"
+```
+
+Import data from Elasticsearch into CrateDB.
+```shell
+ingestr ingest --yes \
+  --source-uri "elasticsearch://elasticsearch:9200?secure=false" \
+  --source-table "example" \
+  --dest-uri "cratedb://crate:crate@cratedb:5432" \
+  --dest-table "doc.example"
 ```
 
 ## Usage

application/ingestr/test.sh

@@ -11,14 +11,6 @@ function setup() {
   if ! command -v uv >/dev/null 2>&1; then
     pip install uv
   fi
-  if ! command -v xonsh >/dev/null 2>&1; then
-    uv tool install xonsh
-  fi
-}
-
-# Invoke Kafka tests.
-function test_kafka() {
-  uvx 'xonsh==0.23.1' kafka-demo.xsh
 }
 
 # Invoke Elasticsearch tests.
@@ -27,5 +19,4 @@ function test_elasticsearch() {
 }
 
 setup
-test_kafka
 test_elasticsearch

application/omniload/.env

@@ -0,0 +1,24 @@
+# Data source configuration (Kafka).
+export PORT_KAFKA_BROKER_EXTERNAL=9092
+export PORT_KAFKA_BROKER_INTERNAL=9094
+export PORT_KAFKA_ZOOKEEPER=2181
+
+# Data sink configuration (CrateDB).
+export CRATEDB_HOST=${CRATEDB_HOST:-cratedb}
+export CRATEDB_HTTP_PORT=${CRATEDB_HTTP_PORT:-4200}
+export CRATEDB_POSTGRESQL_PORT=${CRATEDB_POSTGRESQL_PORT:-5432}
+export CRATEDB_USERNAME=${CRATEDB_USERNAME:-crate}
+export CRATEDB_PASSWORD=${CRATEDB_PASSWORD:-crate}
+export CRATEDB_HTTP_SCHEME=${CRATEDB_HTTP_SCHEME:-http}
+
+# Data sink configuration (CrateDB Cloud).
+# export CRATEDB_HTTP_SCHEME=https
+# export CRATEDB_HOST=example.aks1.westeurope.azure.cratedb.net
+# export CRATEDB_USERNAME='admin'
+# export CRATEDB_PASSWORD='<PASSWORD>'
+
+# Needed by HTTPie.
+export CRATEDB_HTTP_URL_DOCKER="${CRATEDB_HTTP_SCHEME}://${CRATEDB_USERNAME}:${CRATEDB_PASSWORD}@${CRATEDB_HOST}:${CRATEDB_HTTP_PORT}"
+
+# Needed by Python's urllib.
+export CRATEDB_HTTP_URL_NOAUTH="${CRATEDB_HTTP_SCHEME}://localhost:${CRATEDB_HTTP_PORT}"

application/omniload/.gitignore

@@ -0,0 +1,7 @@
+*.json
+*.ndjson
+*.tar.gz
+*.jar
+.venv*
+__pycache__
+!.env

application/omniload/README.md

@@ -0,0 +1,43 @@
+# Use CrateDB with omniload
+
+## About
+
+[omniload] is a command-line application that allows copying data
+from any source into any destination database.
+
+## What's inside
+
+This folder includes runnable examples that use omniload with CrateDB.
+They are also used as integration tests to ensure software components
+fit together well.
+
+- `.env`: Environment variable definitions
+- `kafka-compose.yml`: Service composition file for Kafka and CrateDB
+- `kafka-demo.xsh`: The demonstration program
+
+## Prerequisites
+
+For executing the commands in this walkthrough, you need a working
+installation of Docker and Python.
+
+## Synopsis
+
+```shell
+omniload ingest \
+  --source-uri "kafka://?bootstrap_servers=localhost:9092&group_id=test_group&value_type=json&select=value" \
+  --source-table "demo" \
+  --dest-uri "cratedb://crate:crate@localhost:5432/?sslmode=disable" \
+  --dest-table "doc.kafka_demo"
+```
+
+## Usage
+
+To start cycling without tearing down the backend stack each time,
+use the `KEEPALIVE` environment variable.
+```shell
+export KEEPALIVE=true
+sh test.sh
+```
+
+
+[omniload]: https://omniload.readthedocs.io/

application/ingestr/kafka-compose.yml → application/omniload/kafka-compose.yml

@@ -2,8 +2,8 @@
 # Accompanies `kafka-demo.xsh` and `.env` files.
 ---
 networks:
-  ingestr-demo:
-    name: ingestr-demo
+  omniload-demo:
+    name: omniload-demo
     driver: bridge
 
 services:
@@ -28,7 +28,7 @@ services:
       KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
     networks:
-      - ingestr-demo
+      - omniload-demo
 
     # Define health check for Kafka broker.
     healthcheck:
@@ -55,7 +55,7 @@ services:
       "-Ccluster.routing.allocation.disk.threshold_enabled=false",
     ]
     networks:
-      - ingestr-demo
+      - omniload-demo
 
     # Define health check for CrateDB.
     healthcheck:
@@ -73,31 +73,31 @@ services:
   # Create Kafka topic.
   create-topic:
     image: docker.io/apache/kafka:4.2.0
-    networks: [ingestr-demo]
+    networks: [omniload-demo]
     command: /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:${PORT_KAFKA_BROKER_INTERNAL} --create --if-not-exists --replication-factor 1 --partitions 1 --topic demo
     deploy:
       replicas: 0
 
   # Delete Kafka topic.
   delete-topic:
     image: docker.io/apache/kafka:4.2.0
-    networks: [ingestr-demo]
+    networks: [omniload-demo]
     command: /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:${PORT_KAFKA_BROKER_INTERNAL} --delete --if-exists --topic demo
     deploy:
       replicas: 0
 
   # Publish data to Kafka topic.
   publish-topic:
     image: docker.io/apache/kafka:4.2.0
-    networks: [ingestr-demo]
+    networks: [omniload-demo]
     command: /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server kafka:${PORT_KAFKA_BROKER_INTERNAL} --topic demo
     deploy:
       replicas: 0
 
   # Subscribe to Kafka topic.
   subscribe-topic:
     image: docker.io/edenhill/kcat:1.7.1
-    networks: [ingestr-demo]
+    networks: [omniload-demo]
     command: kcat -b kafka -C -t demo  # -o end
     deploy:
       replicas: 0
@@ -109,22 +109,22 @@ services:
   # Create database table in CrateDB.
   create-table:
     image: docker.io/westonsteimel/httpie
-    networks: [ingestr-demo]
+    networks: [omniload-demo]
     command: http "${CRATEDB_HTTP_URL_DOCKER}/_sql?pretty" stmt='CREATE TABLE "kafka_demo" ("payload" OBJECT(DYNAMIC))' --ignore-stdin
     deploy:
       replicas: 0
 
   # Drop database table in CrateDB.
   drop-table:
     image: docker.io/westonsteimel/httpie
-    networks: [ingestr-demo]
+    networks: [omniload-demo]
     command: http "${CRATEDB_HTTP_URL_DOCKER}/_sql?pretty" stmt='DROP TABLE IF EXISTS "kafka_demo"' --ignore-stdin
     deploy:
       replicas: 0
 
   # Invoke HTTPie via Docker.
   httpie:
     image: docker.io/westonsteimel/httpie
-    networks: [ingestr-demo]
+    networks: [omniload-demo]
     deploy:
       replicas: 0

application/ingestr/kafka-demo.xsh → application/omniload/kafka-demo.xsh

@@ -4,12 +4,11 @@
 # Accompanies `kafka-compose.yml` and `.env` files.
 
 # End-to-end test feeding data through a pipeline implemented with Apache Kafka,
-# ingestr, and CrateDB. The data source is a file in NDJSON format, the
+# omniload, and CrateDB. The data source is a file in NDJSON format, the
 # data sink is a database table in CrateDB.
 
 
 $XONSH_SHOW_TRACEBACK = True
-$INGESTR_DISABLE_TELEMETRY = True
 
 
 # Compute path to directory of current program.
@@ -122,19 +121,18 @@ class Datawrapper:
     def load(self):
         """Load data into CrateDB"""
 
-        # Invoke ingestr job.
-        title "Invoking ingestr job"
+        # Invoke omniload job.
+        title "Invoking omniload job"
 
         result = ![
-            uvx --python=3.12 --prerelease=allow --with-requirements=kafka-requirements.txt \
-                ingestr ingest --yes \
+            omniload ingest \
                 --source-uri "kafka://?bootstrap_servers=localhost:9092&group_id=test_group&value_type=json&select=value" \
                 --source-table "demo" \
                 --dest-uri "cratedb://crate:crate@localhost:5432/?sslmode=disable" \
                 --dest-table "doc.kafka_demo"
         ]
         if result.returncode != 0:
-            echo "ERROR: Failed to ingest data using ingestr"
+            echo "ERROR: Failed to ingest data using omniload"
             exit 1
         echo "Done."
         return self

application/omniload/requirements.txt

@@ -0,0 +1 @@
+omniload<0.3

application/omniload/test.sh

@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+# omniload integration test recipe.
+set -eu
+uvx 'xonsh==0.23.1' kafka-demo.xsh