Skip to content

Commit 3fd6c86

Browse files
amotlamotl
committed
Dask Distributed: Basic remoting / serverless function example
1 parent dd4a132 commit 3fd6c86

7 files changed

Lines changed: 232 additions & 0 deletions

File tree

‎.github/dependabot.yml‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -186,6 +186,22 @@ updates:
186186

187187
# Frameworks.
188188

189+
- directory: "/framework/dask-distributed"
190+
package-ecosystem: "pip"
191+
versioning-strategy: "widen"
192+
schedule:
193+
interval: "daily"
194+
195+
- directory: "/framework/dask-distributed"
196+
package-ecosystem: "docker"
197+
schedule:
198+
interval: "daily"
199+
200+
- directory: "/framework/dask-distributed"
201+
package-ecosystem: "docker-compose"
202+
schedule:
203+
interval: "daily"
204+
189205
- directory: "/framework/dbt/basic"
190206
package-ecosystem: "pip"
191207
versioning-strategy: "widen"

‎.github/workflows/framework-dask-distributed.yml‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
name: "Dask Distributed"
2+
3+
on:
4+
pull_request:
5+
paths:
6+
- '.github/workflows/framework-dask-distributed.yml'
7+
- 'framework/dask-distributed/**'
8+
push:
9+
branches: [ main ]
10+
paths:
11+
- '.github/workflows/framework-dask-distributed.yml'
12+
- 'framework/dask-distributed/**'
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 4 * * *'
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: false
33+
matrix:
34+
os:
35+
- "ubuntu-latest"
36+
cratedb-version:
37+
- 'nightly'
38+
39+
env:
40+
CRATEDB_VERSION: ${{ matrix.cratedb-version }}
41+
42+
name: "
43+
CrateDB ${{ matrix.cratedb-version }}
44+
"
45+
steps:
46+
47+
- name: Acquire sources
48+
uses: actions/checkout@v6
49+
50+
- name: Validate framework/dask-distributed
51+
run: |
52+
# TODO: Generalize invocation into `ngr` test runner.
53+
cd framework/dask-distributed
54+
bash test.sh

‎framework/dask-distributed/Dockerfile‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
ARG DASK_VERSION="latest-py3.13"
2+
3+
FROM docker.io/daskdev/dask:${DASK_VERSION}
4+
5+
# Enable concurrent multi-arch builds.
6+
# https://github.com/docker/buildx/issues/549#issuecomment-1788297892
7+
ARG TARGETPLATFORM
8+
ARG TARGETOS
9+
ARG TARGETARCH
10+
ARG TARGETVARIANT
11+
12+
# Install and configure build prerequisites.
13+
RUN apt-get update && apt-get install --yes git
14+
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
15+
ENV UV_COMPILE_BYTECODE="true"
16+
ENV UV_LINK_MODE="copy"
17+
ENV UV_SYSTEM_PYTHON="true"
18+
19+
# Install payload package.
20+
ADD requirements.txt /
21+
RUN \
22+
--mount=type="cache",id="uv-${TARGETARCH}${TARGETVARIANT}",target="/root/.cache/uv" \
23+
uv pip install --requirement="requirements.txt"

‎framework/dask-distributed/README.md‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
# Using CrateDB with Dask Distributed
2+
3+
Spin up infrastructure.
4+
```shell
5+
export BUILDKIT_PROGRESS=plain
6+
docker compose up --detach --wait
7+
```
8+
9+
Invoke I/O payload.
10+
```shell
11+
docker compose run --rm export-to-deltalake
12+
```
13+
14+
Verify outcome.
15+
```shell
16+
docker compose run --rm verify
17+
```
18+
19+
Watch log files.
20+
```shell
21+
docker compose logs -f
22+
```

‎framework/dask-distributed/compose.yml‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
services:
2+
3+
cratedb:
4+
image: docker.io/crate/crate:${CRATEDB_VERSION:-latest}
5+
command: >
6+
crate
7+
'-Cdiscovery.type=single-node'
8+
'-Cstats.enabled=true'
9+
'-Ccluster.routing.allocation.disk.threshold_enabled=false'
10+
ports:
11+
- "4200:4200"
12+
- "5432:5432"
13+
healthcheck:
14+
test: ["CMD", "curl", "--fail", "http://localhost:4200"]
15+
start_period: 3s
16+
interval: 1.5s
17+
retries: 30
18+
timeout: 30s
19+
20+
dask-scheduler:
21+
build:
22+
context: .
23+
command: dask scheduler --dashboard
24+
ports:
25+
- "8786:8786"
26+
- "8787:8787"
27+
healthcheck:
28+
test: ["CMD", "wget", "--quiet", "-O", "-", "http://localhost:8787/health"]
29+
start_period: 3s
30+
interval: 1.5s
31+
retries: 30
32+
timeout: 30s
33+
34+
dask-worker:
35+
build:
36+
context: .
37+
command: dask worker --dashboard --dashboard-address ":8789"
38+
environment:
39+
- DASK_SCHEDULER_ADDRESS=tcp://dask-scheduler:8786
40+
depends_on:
41+
- dask-scheduler
42+
healthcheck:
43+
test: ["CMD", "wget", "--quiet", "-O", "-", "http://localhost:8789/health"]
44+
start_period: 3s
45+
interval: 1.5s
46+
retries: 30
47+
timeout: 30s
48+
49+
export-to-deltalake:
50+
build:
51+
context: .
52+
command: >
53+
ctk --debug save
54+
"crate://crate:crate@cratedb/sys/summits"
55+
"file+deltalake:///var/delta/summits?mode=overwrite"
56+
environment:
57+
- DASK_SCHEDULER_ADDRESS=tcp://dask-scheduler:8786
58+
depends_on:
59+
cratedb:
60+
condition: service_healthy
61+
dask-scheduler:
62+
condition: service_healthy
63+
dask-worker:
64+
condition: service_healthy
65+
profiles:
66+
- tasks
67+
68+
verify:
69+
build:
70+
context: .
71+
entrypoint:
72+
- /bin/sh
73+
- '-c'
74+
command:
75+
- |
76+
python - <<EOF
77+
import os
78+
import sys
79+
from dask.distributed import Client
80+
with Client() as client:
81+
tasks = client.get_task_stream()
82+
if not tasks:
83+
if os.getenv("mode") == "ignore-empty":
84+
sys.exit(0)
85+
raise RuntimeError("No tasks have been submitted")
86+
task = tasks[-1]
87+
assert task["key"].startswith("save_table"), f"Task has wrong key: {task['key']}"
88+
assert task["status"] == "OK", f"Task status not OK: {task['status']}"
89+
print("SUCCESS")
90+
EOF
91+
environment:
92+
- DASK_SCHEDULER_ADDRESS=tcp://dask-scheduler:8786
93+
profiles:
94+
- tasks

‎framework/dask-distributed/requirements.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
cratedb-toolkit[io,io-opentable] @ git+https://github.com/crate/cratedb-toolkit@io-dask

‎framework/dask-distributed/test.sh‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
#!/bin/sh
2+
3+
# Use Dask with CrateDB.
4+
5+
# The miniature stack defines {Docker,Podman} services and tasks to spin
6+
# up CrateDB and a Dask cluster, and run an I/O job on it.
7+
8+
# https://github.com/crate/crate
9+
# https://github.com/dask/distributed
10+
11+
# Start services.
12+
export BUILDKIT_PROGRESS=plain
13+
docker compose up --detach --wait
14+
15+
# Pre-flight request to fix Dask accounting.
16+
docker compose run --rm --env="mode=ignore-empty" verify
17+
18+
# Run I/O example.
19+
docker compose run --rm export-to-deltalake
20+
21+
# Verify task was invoked properly.
22+
docker compose run --rm verify

0 commit comments

Comments
 (0)