Skip to content

Commit f24449e

Browse files
amotlamotl
committed
I/O: Migrate Kafka->CrateDB recipe from ingestr to omniload
ingestr v1 migrated to Go, but omniload inherited the Python codebase. It also integrated the long-awaited pending patch to improve the Kafka decoder, so everything we need to transfer data from Kafka to CrateDB is nicely wrapped into a package going forward.
1 parent 7048895 commit f24449e

14 files changed

Lines changed: 180 additions & 41 deletions

File tree

‎.github/dependabot.yml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,16 @@ updates:
173173
schedule:
174174
interval: "daily"
175175

176+
- directory: "/application/omniload"
177+
package-ecosystem: "pip"
178+
schedule:
179+
interval: "daily"
180+
181+
- directory: "/application/omniload"
182+
package-ecosystem: "docker-compose"
183+
schedule:
184+
interval: "daily"
185+
176186
- directory: "/application/pgpool"
177187
package-ecosystem: "docker-compose"
178188
schedule:

‎.github/workflows/application-ingestr.yml‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
name: "Ingestr"
1+
name: "ingestr"
22

33
on:
44
pull_request:
@@ -40,7 +40,11 @@ jobs:
4040
uses: actions/checkout@v6
4141

4242
- name: Install uv
43-
uses: astral-sh/setup-uv@v7
43+
uses: astral-sh/setup-uv@v8.2.0
44+
with:
45+
activate-environment: 'true'
46+
enable-cache: true
47+
python-version: '3.14'
4448

4549
- name: Validate application/ingestr
4650
run: |

‎.github/workflows/application-omniload.yml‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
name: "omniload"
2+
3+
on:
4+
pull_request:
5+
paths:
6+
- '.github/workflows/application-omniload.yml'
7+
- 'application/omniload/**'
8+
push:
9+
branches: [ main ]
10+
paths:
11+
- '.github/workflows/application-omniload.yml'
12+
- 'application/omniload/**'
13+
14+
# Allow job to be triggered manually.
15+
workflow_dispatch:
16+
17+
# Run job each night after CrateDB nightly has been published.
18+
schedule:
19+
- cron: '0 3 * * *'
20+
21+
# Cancel in-progress jobs when pushing to the same branch.
22+
concurrency:
23+
cancel-in-progress: true
24+
group: ${{ github.workflow }}-${{ github.ref }}
25+
26+
jobs:
27+
28+
test:
29+
runs-on: ${{ matrix.os }}
30+
31+
strategy:
32+
fail-fast: true
33+
matrix:
34+
os: [ "ubuntu-latest" ]
35+
36+
name: OS ${{ matrix.os }}
37+
steps:
38+
39+
- name: Acquire sources
40+
uses: actions/checkout@v6
41+
42+
- name: Install uv
43+
uses: astral-sh/setup-uv@v8.2.0
44+
with:
45+
activate-environment: 'true'
46+
enable-cache: true
47+
python-version: '3.14'
48+
49+
- name: Validate application/omniload
50+
run: |
51+
cd application/omniload
52+
uv pip install -r requirements.txt
53+
sh test.sh

‎application/ingestr/.env‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,3 @@
1-
# Data source configuration (Kafka).
2-
export PORT_KAFKA_BROKER_EXTERNAL=9092
3-
export PORT_KAFKA_BROKER_INTERNAL=9094
4-
export PORT_KAFKA_ZOOKEEPER=2181
5-
61
# Data sink configuration (CrateDB).
72
export CRATEDB_HOST=${CRATEDB_HOST:-cratedb}
83
export CRATEDB_HTTP_PORT=${CRATEDB_HTTP_PORT:-4200}

‎application/ingestr/README.md‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,8 @@ They are also used as integration tests to ensure software components
1212
fit together well.
1313

1414
- `.env`: Environment variable definitions
15-
- `kafka-compose.yml`: Service composition file for Kafka and CrateDB
16-
- `kafka-demo.xsh`: The demonstration program
15+
- `elasticsearch-compose.yml`: Service composition file for Elasticsearch and CrateDB
16+
- `elasticsearch-demo.sh`: The demonstration program
1717

1818
## Prerequisites
1919

@@ -22,12 +22,22 @@ installation of Docker and Python.
2222

2323
## Synopsis
2424

25+
Import data from CSV into Elasticsearch.
2526
```shell
2627
ingestr ingest --yes \
27-
--source-uri "kafka://?bootstrap_servers=localhost:9092&group_id=test_group&value_type=json&select=value" \
28-
--source-table "demo" \
29-
--dest-uri "cratedb://crate:crate@localhost:5432/?sslmode=disable" \
30-
--dest-table "doc.kafka_demo"
28+
--source-uri "csv://taxi_details.csv" \
29+
--source-table "data" \
30+
--dest-uri "elasticsearch://elasticsearch:9200?secure=false" \
31+
--dest-table "example"
32+
```
33+
34+
Import data from Elasticsearch into CrateDB.
35+
```shell
36+
ingestr ingest --yes \
37+
--source-uri "elasticsearch://elasticsearch:9200?secure=false" \
38+
--source-table "example" \
39+
--dest-uri "cratedb://crate:crate@cratedb:5432" \
40+
--dest-table "doc.example"
3141
```
3242

3343
## Usage

‎application/ingestr/kafka-requirements.txt‎

Lines changed: 0 additions & 1 deletion
This file was deleted.

‎application/ingestr/test.sh‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -11,14 +11,6 @@ function setup() {
1111
if ! command -v uv >/dev/null 2>&1; then
1212
pip install uv
1313
fi
14-
if ! command -v xonsh >/dev/null 2>&1; then
15-
uv tool install xonsh
16-
fi
17-
}
18-
19-
# Invoke Kafka tests.
20-
function test_kafka() {
21-
uvx 'xonsh==0.23.1' kafka-demo.xsh
2214
}
2315

2416
# Invoke Elasticsearch tests.
@@ -27,5 +19,4 @@ function test_elasticsearch() {
2719
}
2820

2921
setup
30-
test_kafka
3122
test_elasticsearch

‎application/omniload/.env‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
# Data source configuration (Kafka).
2+
export PORT_KAFKA_BROKER_EXTERNAL=9092
3+
export PORT_KAFKA_BROKER_INTERNAL=9094
4+
export PORT_KAFKA_ZOOKEEPER=2181
5+
6+
# Data sink configuration (CrateDB).
7+
export CRATEDB_HOST=${CRATEDB_HOST:-cratedb}
8+
export CRATEDB_HTTP_PORT=${CRATEDB_HTTP_PORT:-4200}
9+
export CRATEDB_POSTGRESQL_PORT=${CRATEDB_POSTGRESQL_PORT:-5432}
10+
export CRATEDB_USERNAME=${CRATEDB_USERNAME:-crate}
11+
export CRATEDB_PASSWORD=${CRATEDB_PASSWORD:-crate}
12+
export CRATEDB_HTTP_SCHEME=${CRATEDB_HTTP_SCHEME:-http}
13+
14+
# Data sink configuration (CrateDB Cloud).
15+
# export CRATEDB_HTTP_SCHEME=https
16+
# export CRATEDB_HOST=example.aks1.westeurope.azure.cratedb.net
17+
# export CRATEDB_USERNAME='admin'
18+
# export CRATEDB_PASSWORD='<PASSWORD>'
19+
20+
# Needed by HTTPie.
21+
export CRATEDB_HTTP_URL_DOCKER="${CRATEDB_HTTP_SCHEME}://${CRATEDB_USERNAME}:${CRATEDB_PASSWORD}@${CRATEDB_HOST}:${CRATEDB_HTTP_PORT}"
22+
23+
# Needed by Python's urllib.
24+
export CRATEDB_HTTP_URL_NOAUTH="${CRATEDB_HTTP_SCHEME}://localhost:${CRATEDB_HTTP_PORT}"

‎application/omniload/.gitignore‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
*.json
2+
*.ndjson
3+
*.tar.gz
4+
*.jar
5+
.venv*
6+
__pycache__
7+
!.env

‎application/omniload/README.md‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
# Use CrateDB with omniload
2+
3+
## About
4+
5+
[omniload] is a command-line application that allows copying data
6+
from any source into any destination database.
7+
8+
## What's inside
9+
10+
This folder includes runnable examples that use omniload with CrateDB.
11+
They are also used as integration tests to ensure software components
12+
fit together well.
13+
14+
- `.env`: Environment variable definitions
15+
- `kafka-compose.yml`: Service composition file for Kafka and CrateDB
16+
- `kafka-demo.xsh`: The demonstration program
17+
18+
## Prerequisites
19+
20+
For executing the commands in this walkthrough, you need a working
21+
installation of Docker and Python.
22+
23+
## Synopsis
24+
25+
```shell
26+
omniload ingest \
27+
--source-uri "kafka://?bootstrap_servers=localhost:9092&group_id=test_group&value_type=json&select=value" \
28+
--source-table "demo" \
29+
--dest-uri "cratedb://crate:crate@localhost:5432/?sslmode=disable" \
30+
--dest-table "doc.kafka_demo"
31+
```
32+
33+
## Usage
34+
35+
To start cycling without tearing down the backend stack each time,
36+
use the `KEEPALIVE` environment variable.
37+
```shell
38+
export KEEPALIVE=true
39+
sh test.sh
40+
```
41+
42+
43+
[omniload]: https://omniload.readthedocs.io/

0 commit comments

Comments
 (0)