Dask Distributed: Basic remoting / serverless function example

.github/dependabot.yml

@@ -186,6 +186,22 @@ updates:
 
   # Frameworks.
 
+  - directory: "/framework/dask-distributed"
+    package-ecosystem: "pip"
+    versioning-strategy: "widen"
+    schedule:
+      interval: "daily"
+
+  - directory: "/framework/dask-distributed"
+    package-ecosystem: "docker"
+    schedule:
+      interval: "daily"
+
+  - directory: "/framework/dask-distributed"
+    package-ecosystem: "docker-compose"
+    schedule:
+      interval: "daily"
+
   - directory: "/framework/dbt/basic"
     package-ecosystem: "pip"
     versioning-strategy: "widen"

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

@@ -0,0 +1,54 @@
+name: "Dask Distributed"
+
+on:
+  pull_request:
+    paths:
+    - '.github/workflows/framework-dask-distributed.yml'
+    - 'framework/dask-distributed/**'
+  push:
+    branches: [ main ]
+    paths:
+    - '.github/workflows/framework-dask-distributed.yml'
+    - 'framework/dask-distributed/**'
+
+  # Allow job to be triggered manually.
+  workflow_dispatch:
+
+  # Run job each night after CrateDB nightly has been published.
+  schedule:
+    - cron: '0 4 * * *'
+
+# 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: false
+      matrix:
+        os:
+          - "ubuntu-latest"
+        cratedb-version:
+          - 'nightly'
+
+    env:
+      CRATEDB_VERSION: ${{ matrix.cratedb-version }}
+
+    name: "
+    CrateDB ${{ matrix.cratedb-version }}
+    "
+    steps:
+
+      - name: Acquire sources
+        uses: actions/checkout@v6
+
+      - name: Validate framework/dask-distributed
+        run: |
+          # TODO: Generalize invocation into `ngr` test runner.
+          cd framework/dask-distributed
+          bash test.sh

framework/dask-distributed/Dockerfile

@@ -0,0 +1,23 @@
+ARG DASK_VERSION="latest-py3.13"
+
+FROM docker.io/daskdev/dask:${DASK_VERSION}
+
+# Enable concurrent multi-arch builds.
+# https://github.com/docker/buildx/issues/549#issuecomment-1788297892
+ARG TARGETPLATFORM
+ARG TARGETOS
+ARG TARGETARCH
+ARG TARGETVARIANT
+
+# Install and configure build prerequisites.
+RUN apt-get update && apt-get install --yes --no-install-recommends --no-install-suggests git
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
+ENV UV_COMPILE_BYTECODE="true"
+ENV UV_LINK_MODE="copy"
+ENV UV_SYSTEM_PYTHON="true"
+
+# Install payload package.
+COPY requirements.txt /
+RUN \
+    --mount=type=cache,id="uv-${TARGETARCH}${TARGETVARIANT}",target="/root/.cache/uv" \
+    uv pip install --requirement="requirements.txt"

framework/dask-distributed/README.md

@@ -0,0 +1,27 @@
+# Using CrateDB with Dask Distributed
+
+Spin up infrastructure.
+```shell
+export BUILDKIT_PROGRESS=plain
+docker compose up --detach --wait
+```
+
+Invoke I/O payload.
+```shell
+docker compose run --rm export-to-deltalake
+```
+
+Verify outcome.
+```shell
+docker compose run --rm verify
+```
+
+Watch log files.
+```shell
+docker compose logs -f
+```
+
+Navigate to the Dask dashboard.
+```shell
+open http://localhost:8787/
+```

framework/dask-distributed/compose.yml

@@ -0,0 +1,88 @@
+services:
+
+  cratedb:
+    image: docker.io/crate/crate:${CRATEDB_VERSION:-latest}
+    command: >
+      crate
+        '-Cdiscovery.type=single-node'
+        '-Cstats.enabled=true'
+    ports:
+      - "4200:4200"
+      - "5432:5432"
+    healthcheck:
+      test: ["CMD", "curl", "--fail", "http://localhost:4200"]
+      start_period: 3s
+      interval: 1.5s
+      retries: 30
+      timeout: 30s
+
+  dask-scheduler:
+    build:
+      context: .
+    command: dask scheduler --dashboard
+    ports:
+      - "8786:8786"
+      - "8787:8787"
+    healthcheck:
+      test: ["CMD", "wget", "--quiet", "-O", "-", "http://localhost:8787/health"]
+      start_period: 3s
+      interval: 1.5s
+      retries: 30
+      timeout: 30s
+
+  dask-worker:
+    build:
+      context: .
+    command: dask worker --dashboard --dashboard-address ":8789"
+    environment:
+      DASK_SCHEDULER_ADDRESS: "tcp://dask-scheduler:8786"
+    depends_on:
+      dask-scheduler:
+        condition: service_healthy
+    healthcheck:
+      test: ["CMD", "wget", "--quiet", "-O", "-", "http://localhost:8789/health"]
+      start_period: 3s
+      interval: 1.5s
+      retries: 30
+      timeout: 30s
+
+  export-to-deltalake:
+    build:
+      context: .
+    command: >
+      ctk --debug save
+        "crate://crate:crate@cratedb/sys/summits"
+        "file+deltalake:///var/delta/summits?mode=overwrite"
+    environment:
+      DASK_SCHEDULER_ADDRESS: "tcp://dask-scheduler:8786"
+    profiles:
+      - tasks
+
+  verify:
+    build:
+      context: .
+    entrypoint:
+      - /bin/sh
+      - '-c'
+    command:
+      - |
+        python - <<EOF
+        import os
+        import sys
+        from dask.distributed import Client
+        address = os.environ["DASK_SCHEDULER_ADDRESS"]
+        with Client(address) as client:
+          tasks = client.get_task_stream()
+          if os.getenv("mode") == "ping":
+            sys.exit(0)
+          if not tasks:
+            raise RuntimeError("No tasks have been submitted")
+          task = tasks[-1]
+          assert task["key"].startswith("save_table"), f"Task has wrong key: {task['key']}"
+          assert task["status"] == "OK", f"Task status not OK: {task['status']}"
+          print("SUCCESS")
+        EOF
+    environment:
+      DASK_SCHEDULER_ADDRESS: "tcp://dask-scheduler:8786"
+    profiles:
+      - tasks

framework/dask-distributed/requirements.txt

@@ -0,0 +1 @@
+cratedb-toolkit[io,io-opentable]>=0.0.48

framework/dask-distributed/test.sh

@@ -0,0 +1,24 @@
+#!/bin/sh
+set -e
+
+# Use Dask with CrateDB.
+
+# The miniature stack defines {Docker,Podman} services and tasks to spin
+# up CrateDB and a Dask cluster, and run an I/O job on it.
+
+# https://github.com/crate/crate
+# https://github.com/dask/distributed
+
+# Start services.
+export BUILDKIT_PROGRESS=plain
+docker compose up --detach --wait
+
+# Pre-flight request to fix Dask accounting.
+# https://github.com/dask/distributed/issues/9253
+docker compose run --rm --quiet-build --env="mode=ping" verify
+
+# Run I/O example.
+docker compose run --rm --quiet-build export-to-deltalake
+
+# Verify task was invoked properly.
+docker compose run --rm --quiet-build verify