This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark-connect-rust.git
The following commit(s) were added to refs/heads/master by this push:
new 9d01795 [SPARK-58983][INFRA] Fix CI so all workflows run and pass
9d01795 is described below
commit 9d0179592039d5e4f7a53599c5de17b388c83659
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Tue Aug 25 16:21:10 2026 +0900
[SPARK-58983][INFRA] Fix CI so all workflows run and pass
### What changes were proposed in this pull request?
This makes CI on apache/spark-connect-rust actually **run and go green**.
It started as
"replace non-allowlisted GitHub Actions" (every workflow was failing at
`startup_failure`);
once the workflows could start, several genuine failures surfaced that the
default build
never exercised, and this PR fixes those too.
**1. Use only ASF-allowlisted actions (fixes `startup_failure`).** Replace
third-party
actions across all workflows with GitHub-authored `actions/*` + shell:
- `actions-rust-lang/setup-rust-toolchain`, `dtolnay/rust-toolchain` →
`rustup`
(preinstalled on the runners) + `actions/cachev4` for the cargo/`target`
cache.
- `arduino/setup-protoc` → `protobuf-compiler` via `apt`/`brew`/`choco`
(per-OS).
- `PyO3/maturin-action` → `pip install maturin && maturin build`.
- `pypa/gh-action-pypi-publish` → `twine upload`.
**2. Fix a pre-existing crate-name bug** in `build_python_connect.yml`'s
`rust-checks` job,
which ran `cargo test -p spark-connect -p spark-connect-core` using the old
crate names
(renamed to `apache-spark-connect` / `apache-spark-connect-core`), so it
could never pass.
**3. Fix a `wasm-udf` compile error:** a unit test named
`parse_python_version_output`
shadowed the module function of the same name (imported via `use
super::*`); renamed the
test to `parse_python_version_output_trims_whitespace` so the `wasm-udf`
feature compiles.
**4. Run the WASM packer against the reference pyspark.** The `wasm-udf`
"Packer
end-to-end" step pointed `SPARK_CONNECT_WASM_PACKER_PATH` at `./python`,
which put the
Rust-backed `pyspark` skin on the packer's import path.
`pyspark_wasm_udf.pack` needs
`pyspark.cloudpickle` and `pyspark.sql.types._parse_datatype_json_value`,
and must emit a
reference `DataType` the executors unpickle — none of which the skin
provides. Isolate
`pyspark_wasm_udf` in a `.packer` dir so `import pyspark` resolves to the
reference pyspark
installed earlier in the job.
**5. Unblock the polars build.** The datafusion/polars conversion build
failed on the CI's
stable rustc: the transitive dep `ethnum 1.5.0` does `mem::transmute::<(),
TryFromIntError>(())`,
which no longer compiles now that std's `TryFromIntError` is not zero-sized
(E0512: source
`()` = 0 bits, target `TryFromIntError` = 8 bits). Bump `ethnum` to `1.5.3`
(still within
polars' `^1.5` requirement).
**6. Move the example crates and datasets under `examples/`**
(`examples/wasm-udfs`,
`examples/wasm-udf-inline`, `examples/datasets`) and update the workflow
paths accordingly;
also drop the unused `docker-compose.yml` / `pre-commit.sh` /
`.pre-commit-config.yaml`.
**7. Fix a transport-injection stub hang that timed out the Connect parity
gate.** On
`test_client.py`, `test_server_unreachable` builds a client to an
unresolvable host
(`sc://foo`) with the Python retry policy set to `max_retries=0` and
expects a prompt
`UNAVAILABLE`. Our `RustConnectStub` (used only by the parity harness's
`rust_transport_plugin`) ran RPCs through the Rust client's `with_retry`
using
`RetryPolicy::default()` (15 retries, backoff to 64s) — but the stub is a
drop-in for the
single-shot grpcio stub, over which the reference client layers its *own*
retry handler.
Retrying inside our client double-retried and ignored the client-configured
policy, turning
a fast failure into a ~700s hang (`ours(p=0,f=0) TIMEOUT`). Fix: add
`RetryPolicy::no_retries()` and set it on the injection stub so RPCs are
single-shot and the
reference client owns retrying (real drop-in client keeps
`RetryPolicy::default()`); also
bound `grpc.ready()` and add `connect_timeout`, surfacing an unreachable
channel as
`UNAVAILABLE` instead of waiting on a lazy channel that never connects.
`test_client.py` now
passes 41/41 through our transport, matching the reference.
### Why are the changes needed?
Every workflow failed with `startup_failure` on pushes to `master` and on
pull requests, so
the repository had no working CI. The workflow YAML is valid; the ASF
GitHub org restricts
Actions to `actions/*` plus a vetted allowlist, and these workflows
referenced third-party
actions that aren't on it, so the runs never started. Once that was fixed,
items 2–5 are the
concrete build/test failures that had never run before — each would
otherwise keep CI red.
### Does this PR introduce any user-facing change?
No — CI/infra and repo layout only.
### How was this patch tested?
CI on this PR itself is the end-to-end validation. Locally: the WASM packer
end-to-end test
(`packer_produces_nonempty_command`) passes against the reference pyspark
with
`pyspark_wasm_udf` isolated on the packer path; `actions/*`-only was
verified (a `grep` for
non-`actions/*` `uses:` is empty) and all workflow files parse as valid
YAML.
JIRA: https://issues.apache.org/jira/browse/SPARK-58983
This pull request and its description were written by Isaac.
Closes #52 from HyukjinKwon/fix-ci-actions.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.github/workflows/build_python_connect.yml | 52 ++++--
.github/workflows/lint.yml | 20 ++-
.github/workflows/release.yml | 52 +++---
.github/workflows/rust.yml | 54 +++++--
.gitignore | 3 +
.isaac/config.json | 3 +
.pre-commit-config.yaml | 42 -----
Cargo.lock | 4 +-
Cargo.toml | 15 +-
README.md | 31 ++--
crates/pyspark-rs/src/transport.rs | 9 +-
crates/spark-connect-core/src/client.rs | 45 ++++--
crates/spark-connect-core/src/retries.rs | 13 ++
crates/spark-connect/src/wasm_udf.rs | 6 +-
docker-compose.yml | 33 ----
examples/Cargo.toml | 2 +-
examples/README.md | 6 +-
.../datasets}/dir1/dir2/file2.parquet | Bin
{datasets => examples/datasets}/dir1/file1.parquet | Bin
{datasets => examples/datasets}/dir1/file3.json | 0
{datasets => examples/datasets}/employees.json | 0
{datasets => examples/datasets}/full_user.avsc | 0
{datasets => examples/datasets}/kv1.txt | 0
{datasets => examples/datasets}/people.csv | 0
{datasets => examples/datasets}/people.json | 0
{datasets => examples/datasets}/people.txt | 0
{datasets => examples/datasets}/user.avsc | 0
{datasets => examples/datasets}/users.avro | Bin
{datasets => examples/datasets}/users.orc | Bin
{datasets => examples/datasets}/users.parquet | Bin
.../wasm-udf-inline}/Cargo.toml | 14 +-
.../wasm-udf-inline}/build.rs | 0
.../wasm-udf-inline}/src/main.rs | 0
{wasm-udfs => examples/wasm-udfs}/Cargo.toml | 12 +-
{wasm-udfs => examples/wasm-udfs}/build.rs | 0
{wasm-udfs => examples/wasm-udfs}/src/lib.rs | 0
pre-commit.sh | 86 ----------
python/pyspark_wasm_udf/pack.py | 8 +-
wasm_udf_example.txt | 175 ---------------------
39 files changed, 237 insertions(+), 448 deletions(-)
diff --git a/.github/workflows/build_python_connect.yml
b/.github/workflows/build_python_connect.yml
index d045828..1f5f1f0 100644
--- a/.github/workflows/build_python_connect.yml
+++ b/.github/workflows/build_python_connect.yml
@@ -64,11 +64,21 @@ jobs:
java-version: "17"
- name: Install Rust toolchain
- uses: actions-rust-lang/setup-rust-toolchain@v1
+ # rustup is preinstalled; avoid third-party toolchain actions (ASF
Actions
+ # policy allowlists actions/* only).
+ run: |
+ rustup toolchain install stable --profile minimal
+ rustup default stable
+
+ - name: Cache cargo & target
+ uses: actions/cache@v4
with:
- toolchain: stable
- rustflags: ""
- cache: true
+ path: |
+ ~/.cargo/registry
+ ~/.cargo/git
+ target
+ key: ${{ runner.os }}-cargo-parity-${{ hashFiles('**/Cargo.lock') }}
+ restore-keys: ${{ runner.os }}-cargo-parity-
- name: Install system dependencies
run: |
@@ -206,11 +216,19 @@ jobs:
with:
distribution: temurin
java-version: "17"
- - uses: actions-rust-lang/setup-rust-toolchain@v1
+ - name: Install Rust toolchain
+ run: |
+ rustup toolchain install stable --profile minimal
+ rustup default stable
+ - name: Cache cargo & target
+ uses: actions/cache@v4
with:
- toolchain: stable
- rustflags: ""
- cache: true
+ path: |
+ ~/.cargo/registry
+ ~/.cargo/git
+ target
+ key: ${{ runner.os }}-cargo-bench-${{ hashFiles('**/Cargo.lock') }}
+ restore-keys: ${{ runner.os }}-cargo-bench-
- name: Install deps
run: |
sudo apt-get update && sudo apt-get install -y protobuf-compiler
@@ -278,17 +296,25 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- - uses: actions-rust-lang/setup-rust-toolchain@v1
+ - name: Install Rust toolchain
+ run: |
+ rustup toolchain install stable --profile minimal
+ rustup default stable
+ - name: Cache cargo & target
+ uses: actions/cache@v4
with:
- toolchain: stable
- rustflags: ""
- cache: true
+ path: |
+ ~/.cargo/registry
+ ~/.cargo/git
+ target
+ key: ${{ runner.os }}-cargo-checks-${{ hashFiles('**/Cargo.lock') }}
+ restore-keys: ${{ runner.os }}-cargo-checks-
- name: Install system dependencies
run: |
sudo apt-get update && sudo apt-get install -y protobuf-compiler
- name: Cargo build
run: cargo build --release
- name: Cargo test
- run: cargo test -p spark-connect -p spark-connect-core --release
+ run: cargo test -p apache-spark-connect -p apache-spark-connect-core
--release
- name: No-stub audit
run: bash scripts/audit_no_stubs.sh
diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
index 422fba0..9569a6c 100644
--- a/.github/workflows/lint.yml
+++ b/.github/workflows/lint.yml
@@ -32,12 +32,22 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- - uses: actions-rust-lang/setup-rust-toolchain@v1
+ - name: Install Rust toolchain (rustfmt + clippy)
+ # rustup is preinstalled; avoid third-party toolchain actions (ASF
Actions
+ # policy allowlists actions/* only).
+ run: |
+ rustup toolchain install stable --profile minimal
+ rustup default stable
+ rustup component add rustfmt clippy
+ - name: Cache cargo & target
+ uses: actions/cache@v4
with:
- toolchain: stable
- rustflags: ""
- components: rustfmt, clippy
- cache: true
+ path: |
+ ~/.cargo/registry
+ ~/.cargo/git
+ target
+ key: ${{ runner.os }}-cargo-lint-${{ hashFiles('**/Cargo.lock') }}
+ restore-keys: ${{ runner.os }}-cargo-lint-
- name: Install system dependencies
run: |
sudo apt-get update && sudo apt-get install -y protobuf-compiler
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index d6d0288..0f99673 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -78,12 +78,12 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- - uses: dtolnay/rust-toolchain@stable
+ - name: Install Rust toolchain
+ run: |
+ rustup toolchain install stable --profile minimal
+ rustup default stable
- name: Install protoc
- uses: arduino/setup-protoc@v3
- with:
- version: "23.x"
- repo-token: ${{ secrets.GITHUB_TOKEN }}
+ run: sudo apt-get update && sudo apt-get install -y protobuf-compiler
- name: cargo publish (dependency order)
env:
CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
@@ -110,19 +110,24 @@ jobs:
- uses: actions/setup-python@v5
with:
python-version: "3.11"
- - uses: dtolnay/rust-toolchain@stable
+ - name: Install Rust toolchain
+ run: |
+ rustup toolchain install stable --profile minimal
+ rustup default stable
- name: Install protoc
- uses: arduino/setup-protoc@v3
- with:
- version: "23.x"
- repo-token: ${{ secrets.GITHUB_TOKEN }}
+ shell: bash
+ run: |
+ case "$RUNNER_OS" in
+ Linux) sudo apt-get update && sudo apt-get install -y
protobuf-compiler ;;
+ macOS) brew install protobuf ;;
+ Windows) choco install protoc -y ;;
+ esac
- name: Build abi3 wheel with maturin
# abi3-py39 produces a single wheel per platform valid for CPython
3.9+.
- uses: PyO3/maturin-action@v1
- with:
- command: build
- args: --release --out dist
- manylinux: auto
+ shell: bash
+ run: |
+ python -m pip install --upgrade maturin
+ maturin build --release --out dist
- uses: actions/upload-artifact@v4
with:
name: wheels-${{ matrix.os }}
@@ -134,15 +139,20 @@ jobs:
if: github.event_name == 'push' || (github.event_name ==
'workflow_dispatch' && inputs.dry_run == false)
runs-on: ubuntu-latest
steps:
+ - uses: actions/setup-python@v5
+ with:
+ python-version: "3.11"
- uses: actions/download-artifact@v4
with:
pattern: wheels-*
path: dist
merge-multiple: true
- name: Publish to PyPI
- uses: pypa/gh-action-pypi-publish@release/v1
- with:
- packages-dir: dist
- # Fork-based release: authenticate with the committer's PyPI token
- # rather than repository-scoped Trusted Publishing.
- password: ${{ secrets.PYPI_API_TOKEN }}
+ # Fork-based release: authenticate with the committer's PyPI token via
twine
+ # (avoids third-party publish actions; ASF Actions policy allowlists
actions/*).
+ env:
+ TWINE_USERNAME: __token__
+ TWINE_PASSWORD: ${{ secrets.PYPI_API_TOKEN }}
+ run: |
+ python -m pip install --upgrade twine
+ twine upload dist/*
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index d2e4979..bb1d7e0 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -36,11 +36,22 @@ jobs:
- uses: actions/checkout@v4
- name: Install Rust toolchain
- uses: actions-rust-lang/setup-rust-toolchain@v1
+ # rustup is preinstalled on the runner; use it directly so we depend
only on
+ # GitHub-authored actions (ASF Actions policy allowlists actions/*, not
+ # third-party toolchain actions).
+ run: |
+ rustup toolchain install ${{ matrix.rust-version }} --profile minimal
+ rustup default ${{ matrix.rust-version }}
+
+ - name: Cache cargo & target
+ uses: actions/cache@v4
with:
- toolchain: ${{ matrix.rust-version }}
- rustflags: ""
- cache: true
+ path: |
+ ~/.cargo/registry
+ ~/.cargo/git
+ target
+ key: ${{ runner.os }}-cargo-rust-${{ hashFiles('**/Cargo.lock') }}
+ restore-keys: ${{ runner.os }}-cargo-rust-
- name: Install system dependencies
run: |
@@ -74,12 +85,20 @@ jobs:
- uses: actions/checkout@v4
- name: Install Rust toolchain (+ wasm32 target)
- uses: actions-rust-lang/setup-rust-toolchain@v1
+ run: |
+ rustup toolchain install stable --profile minimal
+ rustup default stable
+ rustup target add wasm32-unknown-unknown
+
+ - name: Cache cargo & target
+ uses: actions/cache@v4
with:
- toolchain: stable
- target: wasm32-unknown-unknown
- rustflags: ""
- cache: true
+ path: |
+ ~/.cargo/registry
+ ~/.cargo/git
+ target
+ key: ${{ runner.os }}-cargo-wasm-${{ hashFiles('**/Cargo.lock') }}
+ restore-keys: ${{ runner.os }}-cargo-wasm-
- name: Install system dependencies
run: |
@@ -98,11 +117,20 @@ jobs:
- name: Build the macro/build helpers and WASM UDF crates + examples
run: |
cargo build -p apache-spark-connect-macros -p
apache-spark-connect-build
- cargo build --manifest-path wasm-udfs/Cargo.toml
- cargo build --manifest-path wasm-udf-inline/Cargo.toml
+ cargo build --manifest-path examples/wasm-udfs/Cargo.toml
+ cargo build --manifest-path examples/wasm-udf-inline/Cargo.toml
cargo build -p examples --bin wasm_udf_macro --features
examples/wasm-udf-macro
- name: Packer end-to-end (Rust -> cloudpickle command)
+ # The packer runs `python -m pyspark_wasm_udf.pack`, which imports the
+ # reference `pyspark` (installed above) for `pyspark.cloudpickle` and
+ # `pyspark.sql.types._parse_datatype_json_value`, and produces a
+ # reference `DataType` the executors unpickle. Put *only* the
+ # `pyspark_wasm_udf` package on the packer path (not ./python, whose
+ # `pyspark` skin is Rust-backed and would shadow the reference).
env:
- SPARK_CONNECT_WASM_PACKER_PATH: ${{ github.workspace }}/python
- run: cargo test -p apache-spark-connect --features wasm-udf --
--ignored packer_produces
+ SPARK_CONNECT_WASM_PACKER_PATH: ${{ github.workspace }}/.packer
+ run: |
+ mkdir -p .packer
+ cp -r python/pyspark_wasm_udf .packer/
+ cargo test -p apache-spark-connect --features wasm-udf -- --ignored
packer_produces
diff --git a/.gitignore b/.gitignore
index c59e0e6..d598624 100644
--- a/.gitignore
+++ b/.gitignore
@@ -16,6 +16,9 @@
# under the License.
/target
+# The workspace-excluded WASM example crates build to their own target/ dirs.
+/examples/wasm-udfs/target
+/examples/wasm-udf-inline/target
**/*.rs.bk
Cargo.lock
__pycache__/
diff --git a/.isaac/config.json b/.isaac/config.json
new file mode 100644
index 0000000..93cf0a6
--- /dev/null
+++ b/.isaac/config.json
@@ -0,0 +1,3 @@
+{
+ "sync_reminder_last_shown": "2026-08-25"
+}
\ No newline at end of file
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
deleted file mode 100644
index 7d96998..0000000
--- a/.pre-commit-config.yaml
+++ /dev/null
@@ -1,42 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-repos:
- - repo: https://github.com/pre-commit/pre-commit-hooks
- rev: v4.4.0
- hooks:
- - id: check-byte-order-marker
- - id: check-case-conflict
- - id: check-merge-conflict
- - id: check-symlinks
- - id: check-yaml
- - id: end-of-file-fixer
- - id: mixed-line-ending
- - id: trailing-whitespace
- - repo: https://github.com/doublify/pre-commit-rust
- rev: v1.0
- hooks:
- - id: fmt
- - id: cargo-check
- - id: clippy
- - repo: https://github.com/astral-sh/ruff-pre-commit
- rev: v0.14.0
- hooks:
- - id: ruff-check
- args: [--fix, "scripts/"]
- - id: ruff-format
- args: ["scripts/"]
diff --git a/Cargo.lock b/Cargo.lock
index 661af7c..2e4c3c4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1482,9 +1482,9 @@ dependencies = [
[[package]]
name = "ethnum"
-version = "1.5.0"
+version = "1.5.3"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "b90ca2580b73ab6a1f724b76ca11ab632df820fd6040c336200d2c1df7b3c82c"
+checksum = "40404c3f5f511ec4da6fe866ddf6a717c309fdbb69fbbad7b0f3edab8f2e835f"
[[package]]
name = "event-listener"
diff --git a/Cargo.toml b/Cargo.toml
index 563bd9d..10801e0 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -27,19 +27,20 @@ members = [
"examples",
]
exclude = [
- "wasm-udfs",
- "wasm-udf-inline",
+ "examples/wasm-udfs",
+ "examples/wasm-udf-inline",
]
# `pyspark-rs` (the CPython extension cdylib, built via maturin) and the
# Rust-native `examples` crate both link the pure-Rust core directly; both are
# workspace members but excluded from `default-members` so `cargo test`/`cargo
build`
# stay clean. Build them explicitly (`cargo build -p pyspark-rs` / `-p
examples`).
#
-# The `wasm-udfs` and `wasm-udf-inline` crates are excluded from the workspace
-# entirely because they require the `wasm32-unknown-unknown` target, which may
not
-# be installed on every machine (so `cargo test --workspace` stays green
without it).
-# Build them explicitly with `cargo build --manifest-path
wasm-udfs/Cargo.toml` (etc)
-# in CI or when the target is installed locally.
+# The `examples/wasm-udfs` and `examples/wasm-udf-inline` crates are excluded
from
+# the workspace entirely because they require the `wasm32-unknown-unknown`
target,
+# which may not be installed on every machine (so `cargo test --workspace`
stays
+# green without it). Build them explicitly with
+# `cargo build --manifest-path examples/wasm-udfs/Cargo.toml` (etc) in CI or
when
+# the target is installed locally.
default-members = [
"crates/spark-connect-proto",
"crates/spark-connect-core",
diff --git a/README.md b/README.md
index d3f8d54..ffbfc20 100644
--- a/README.md
+++ b/README.md
@@ -29,14 +29,6 @@ building, transport, and Arrow decoding done in Rust.
version tracks the Spark release it targets (starting at `4.2.0`), so the
version
number tells you which Spark it speaks.
-> **Status: alpha, work in progress.** The core DataFrame/Column/functions API,
-> transport, and Arrow result path work end-to-end against a real Spark Connect
-> server. The crates and wheel carry the version **4.2.0** to track the Apache
Spark
-> release they target — it is a compatibility marker, not a maturity claim;
treat the
-> client itself as alpha. API parity with the reference client is validated by
the
-> golden-proto tests (plan building) and the official Apache Spark connect
test suite
-> (transport + Arrow); see [Continuous integration](#continuous-integration).
-
## Why
The reference Spark Connect client is pure Python: it builds protobuf plans,
@@ -134,21 +126,13 @@ fn main() -> Result<()> {
Python UDFs run as normal. In addition, a Rust function can be compiled to
WebAssembly and shipped as a UDF: the `.wasm` bytes are embedded in a
cloudpickled
Python shim and executed on the worker via `wasmtime`, so no Rust toolchain is
-needed server-side. See `python/pyspark_wasm_udf/` and
`examples/src/wasm_udf.rs`.
+needed server-side. See `python/pyspark_wasm_udf/` and the example crates
+`examples/wasm-udf-inline/` (single-file) and `examples/wasm-udfs/` +
+`examples/src/wasm_udf_macro.rs` (reusable crate).
## Getting started with Spark Connect server
-You can run a Spark Connect server locally for development in two ways:
-
-### Option 1: Docker Compose (recommended)
-
-The repo includes a `docker-compose.yml` that starts a Spark Connect server on
port 15002:
-
-```bash
-docker compose up --build -d
-```
-
-### Option 2: Local Spark distribution
+Run a Spark Connect server locally for development:
1. [Download Spark 4.2.0](https://spark.apache.org/downloads.html) and unzip it
2. Set `SPARK_HOME` to the unzipped directory
@@ -166,8 +150,11 @@ The server listens on `sc://localhost:15002` by default.
### Sample data
-The repo includes sample datasets in `datasets/` (people.csv, employees.json,
kv1.txt, etc.)
-mounted at `/opt/spark/work-dir/datasets` in the Docker container for use in
tests and examples.
+Sample datasets used by the examples live in `examples/datasets/` (people.csv,
+employees.json, kv1.txt, etc.). The reader/deltalake examples load them by a
path
+relative to the **server's** working directory (e.g. `./datasets/people.csv`),
so
+start the Spark Connect server from the `examples/` directory (where
`./datasets`
+resolves), or adjust the path in the example to point at the data.
## Continuous integration
diff --git a/crates/pyspark-rs/src/transport.rs
b/crates/pyspark-rs/src/transport.rs
index 8484a91..169c93a 100644
--- a/crates/pyspark-rs/src/transport.rs
+++ b/crates/pyspark-rs/src/transport.rs
@@ -21,6 +21,7 @@ use tokio::sync::Notify;
use spark_connect_core::channel::ChannelBuilder;
use spark_connect_core::client::SparkConnectClient;
+use spark_connect_core::retries::RetryPolicy;
use spark_connect_core::runtime::block_on;
use spark_connect_proto as proto;
@@ -166,9 +167,15 @@ impl RustConnectStub {
let builder = ChannelBuilder::parse(url).map_err(err)?;
// Release the GIL during the connect handshake so a slow/unresponsive
server
// can't wedge the whole interpreter (incl. signal handling / Ctrl-C).
- let client = py
+ let mut client = py
.detach(|| block_on(SparkConnectClient::connect(&builder)))
.map_err(err)?;
+ // This stub is a drop-in for the grpcio stub: a single RPC per call.
The reference
+ // client wraps calls in its own GrpcRetryHandler (per the client-side
retry
+ // policy), so retrying inside our client too would double-retry and
ignore that
+ // policy - a `max_retries=0` client would otherwise hit our 15-retry
backoff and
+ // hang against an unreachable server instead of failing fast with
UNAVAILABLE.
+ client.set_retry_policy(RetryPolicy::no_retries());
Ok(Self {
client,
cancel: Arc::new(Notify::new()),
diff --git a/crates/spark-connect-core/src/client.rs
b/crates/spark-connect-core/src/client.rs
index 6064b61..70438fa 100644
--- a/crates/spark-connect-core/src/client.rs
+++ b/crates/spark-connect-core/src/client.rs
@@ -23,6 +23,28 @@ use crate::error::{Result, SparkError};
use crate::reattach::ExecutePlanResponseReattachableIterator;
use crate::retries::{RetryPolicy, RetryPolicyState};
+/// How long to wait for the lazy gRPC channel to become ready before
surfacing a
+/// retriable `UNAVAILABLE`. tonic's lazy channel waits indefinitely for an
unreachable
+/// or unresolvable host (unlike grpcio, which fails fast), so without this
bound a call
+/// to a down server would hang forever instead of raising `UNAVAILABLE`.
+const READY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+/// Wait for `grpc` to become ready, bounded by [`READY_TIMEOUT`], mapping any
failure
+/// (or the timeout) to a gRPC `UNAVAILABLE` error. This matches how grpcio
reports an
+/// unreachable server, so the reference client raises
`SparkConnectGrpcException` with
+/// `UNAVAILABLE` rather than hanging on a channel that never connects.
+async fn wait_ready(grpc: &mut tonic::client::Grpc<Channel>) -> Result<()> {
+ match tokio::time::timeout(READY_TIMEOUT, grpc.ready()).await {
+ Ok(Ok(())) => Ok(()),
+ Ok(Err(e)) =>
Err(SparkError::from_grpc_status(tonic::Status::unavailable(
+ format!("channel not ready: {e}"),
+ ))),
+ Err(_elapsed) =>
Err(SparkError::from_grpc_status(tonic::Status::unavailable(
+ "channel not ready: connection timed out",
+ ))),
+ }
+}
+
/// A Spark Connect client for communicating with a remote Spark server.
///
/// Mirrors `pyspark.sql.connect.client.core.SparkConnectClient`.
@@ -71,11 +93,13 @@ impl SparkConnectClient {
.map_err(|e| SparkError::connect_msg(format!("Invalid
endpoint: {}", e)))?
.tls_config(tls_config)
.map_err(|e| SparkError::connect_msg(format!("Failed to
configure TLS: {}", e)))?
+ .connect_timeout(READY_TIMEOUT)
.connect_lazy()
} else {
// Plaintext connection (for localhost development)
tonic::transport::Channel::from_shared(format!("http://{endpoint}";))
.map_err(|e| SparkError::connect_msg(format!("Invalid
endpoint: {}", e)))?
+ .connect_timeout(READY_TIMEOUT)
.connect_lazy()
};
@@ -124,6 +148,15 @@ impl SparkConnectClient {
self.user_id.as_deref()
}
+ /// Override the retry policy applied to this client's RPCs.
+ ///
+ /// The transport-injection stub sets this to [`RetryPolicy::no_retries`]
so it behaves
+ /// like the single-shot grpcio stub the reference client expects (which
does its own
+ /// retrying on top); see that constructor for why double-retrying is
wrong there.
+ pub fn set_retry_policy(&mut self, policy: RetryPolicy) {
+ self.retry_policy = policy;
+ }
+
/// Execute a plan and return a stream of responses.
///
/// Mirrors
`pyspark.sql.connect.client.core.SparkConnectClient.execute_plan`.
@@ -145,9 +178,7 @@ impl SparkConnectClient {
/// the reference client built.
pub async fn execute_plan_raw(&self, request: Vec<u8>) ->
Result<Streaming<Vec<u8>>> {
let mut grpc = tonic::client::Grpc::new(self.channel.clone());
- grpc.ready()
- .await
- .map_err(|e| SparkError::connect_msg(format!("channel not ready:
{e}")))?;
+ wait_ready(&mut grpc).await?;
let mut req = Request::new(request);
self._attach_metadata(&mut req);
let path = tonic::codegen::http::uri::PathAndQuery::from_static(
@@ -163,9 +194,7 @@ impl SparkConnectClient {
/// ReattachExecute, byte-exact passthrough returning a raw response
stream.
pub async fn reattach_execute_raw(&self, request: Vec<u8>) ->
Result<Streaming<Vec<u8>>> {
let mut grpc = tonic::client::Grpc::new(self.channel.clone());
- grpc.ready()
- .await
- .map_err(|e| SparkError::connect_msg(format!("channel not ready:
{e}")))?;
+ wait_ready(&mut grpc).await?;
let mut req = Request::new(request);
self._attach_metadata(&mut req);
let path = tonic::codegen::http::uri::PathAndQuery::from_static(
@@ -215,9 +244,7 @@ impl SparkConnectClient {
/// reference client, which never re-decodes the request it built).
pub async fn analyze_plan_raw(&self, request: Vec<u8>) -> Result<Vec<u8>> {
let mut grpc = tonic::client::Grpc::new(self.channel.clone());
- grpc.ready()
- .await
- .map_err(|e| SparkError::connect_msg(format!("channel not ready:
{e}")))?;
+ wait_ready(&mut grpc).await?;
let mut req = Request::new(request);
self._attach_metadata(&mut req);
let path = tonic::codegen::http::uri::PathAndQuery::from_static(
diff --git a/crates/spark-connect-core/src/retries.rs
b/crates/spark-connect-core/src/retries.rs
index 8c749f7..649d3b1 100644
--- a/crates/spark-connect-core/src/retries.rs
+++ b/crates/spark-connect-core/src/retries.rs
@@ -51,6 +51,19 @@ impl Default for RetryPolicy {
}
impl RetryPolicy {
+ /// A policy that performs a single attempt with no retries.
+ ///
+ /// Used by the transport-injection stub, which is a drop-in for the
grpcio stub: the
+ /// reference client layers its own `GrpcRetryHandler` (configured by the
client-side
+ /// retry policy) on top, so retrying here too would double-retry and
ignore that
+ /// policy - e.g. a `max_retries=0` client would still see our 15-retry
backoff.
+ pub fn no_retries() -> Self {
+ Self {
+ max_retries: Some(0),
+ ..Self::default()
+ }
+ }
+
/// Whether a failed RPC is retryable under this policy.
///
/// Mirrors `pyspark.sql.connect.client.retries.DefaultPolicy.can_retry`:
diff --git a/crates/spark-connect/src/wasm_udf.rs
b/crates/spark-connect/src/wasm_udf.rs
index a4167b4..74d8407 100644
--- a/crates/spark-connect/src/wasm_udf.rs
+++ b/crates/spark-connect/src/wasm_udf.rs
@@ -537,9 +537,11 @@ mod tests {
}
#[test]
- fn parse_python_version_output() {
+ fn parse_python_version_output_trims_whitespace() {
// Test that we correctly parse Python version strings from stdout
- // (with potential trailing whitespace/newline)
+ // (with potential trailing whitespace/newline). Note: this test must
not be
+ // named `parse_python_version_output`, or it would shadow the
function under
+ // test (imported via `use super::*`) and fail to compile.
assert_eq!(
parse_python_version_output("3.11\n"),
Some("3.11".to_string())
diff --git a/docker-compose.yml b/docker-compose.yml
deleted file mode 100644
index 8938665..0000000
--- a/docker-compose.yml
+++ /dev/null
@@ -1,33 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-services:
- spark:
- image: "apache/spark:4.2.0"
- command: >
- /opt/spark/sbin/start-connect-server.sh
- --packages
"org.apache.spark:spark-connect_2.13:4.2.0,io.delta:delta-spark_2.13:4.2.0"
- --conf "spark.driver.extraJavaOptions=-Divy.cache.dir=/tmp
-Divy.home=/tmp"
- --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension"
- --conf
"spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"
- environment:
- - SPARK_NO_DAEMONIZE=true
- ports:
- - "4040:4040"
- - "15002:15002"
- volumes:
- - ./datasets:/opt/spark/work-dir/datasets
diff --git a/examples/Cargo.toml b/examples/Cargo.toml
index 8f0bbba..1c389eb 100644
--- a/examples/Cargo.toml
+++ b/examples/Cargo.toml
@@ -56,7 +56,7 @@ required-features = ["wasm-udf-macro"]
[dependencies]
spark-connect = { package = "apache-spark-connect", path =
"../crates/spark-connect" }
-wasm-udfs = { path = "../wasm-udfs", optional = true }
+wasm-udfs = { path = "wasm-udfs", optional = true }
[features]
# Enable the macro example (`wasm_udf_macro` bin). Requires the
diff --git a/examples/README.md b/examples/README.md
index 35e63a3..e84da40 100644
--- a/examples/README.md
+++ b/examples/README.md
@@ -65,10 +65,8 @@ cargo run --bin readstream
Read a file into a dataframe, save the result as a deltalake table, and append
a new record to the table.
-**Prerequisite** the spark cluster must be started with the deltalake package.
The `docker-compose.yml` provided in the repo has deltalake pre-installed.
-Or if you are running a spark connect server location, run the below scripts
first
-
-If you are running a local spark connect server. The Delta Lake jars need to
be added onto the server before it starts.
+**Prerequisite** the spark cluster must be started with the deltalake package.
+The Delta Lake jars need to be added onto the server before it starts:
```bash
$ $SPARK_HOME/sbin/start-connect-server.sh --packages
"org.apache.spark:spark-connect_2.13:4.2.0,io.delta:delta-spark_2.13:4.2.0" \
diff --git a/datasets/dir1/dir2/file2.parquet
b/examples/datasets/dir1/dir2/file2.parquet
similarity index 100%
rename from datasets/dir1/dir2/file2.parquet
rename to examples/datasets/dir1/dir2/file2.parquet
diff --git a/datasets/dir1/file1.parquet b/examples/datasets/dir1/file1.parquet
similarity index 100%
rename from datasets/dir1/file1.parquet
rename to examples/datasets/dir1/file1.parquet
diff --git a/datasets/dir1/file3.json b/examples/datasets/dir1/file3.json
similarity index 100%
rename from datasets/dir1/file3.json
rename to examples/datasets/dir1/file3.json
diff --git a/datasets/employees.json b/examples/datasets/employees.json
similarity index 100%
rename from datasets/employees.json
rename to examples/datasets/employees.json
diff --git a/datasets/full_user.avsc b/examples/datasets/full_user.avsc
similarity index 100%
rename from datasets/full_user.avsc
rename to examples/datasets/full_user.avsc
diff --git a/datasets/kv1.txt b/examples/datasets/kv1.txt
similarity index 100%
rename from datasets/kv1.txt
rename to examples/datasets/kv1.txt
diff --git a/datasets/people.csv b/examples/datasets/people.csv
similarity index 100%
rename from datasets/people.csv
rename to examples/datasets/people.csv
diff --git a/datasets/people.json b/examples/datasets/people.json
similarity index 100%
rename from datasets/people.json
rename to examples/datasets/people.json
diff --git a/datasets/people.txt b/examples/datasets/people.txt
similarity index 100%
rename from datasets/people.txt
rename to examples/datasets/people.txt
diff --git a/datasets/user.avsc b/examples/datasets/user.avsc
similarity index 100%
rename from datasets/user.avsc
rename to examples/datasets/user.avsc
diff --git a/datasets/users.avro b/examples/datasets/users.avro
similarity index 100%
rename from datasets/users.avro
rename to examples/datasets/users.avro
diff --git a/datasets/users.orc b/examples/datasets/users.orc
similarity index 100%
rename from datasets/users.orc
rename to examples/datasets/users.orc
diff --git a/datasets/users.parquet b/examples/datasets/users.parquet
similarity index 100%
rename from datasets/users.parquet
rename to examples/datasets/users.parquet
diff --git a/wasm-udf-inline/Cargo.toml b/examples/wasm-udf-inline/Cargo.toml
similarity index 73%
rename from wasm-udf-inline/Cargo.toml
rename to examples/wasm-udf-inline/Cargo.toml
index 4ff7f75..68f5dcd 100644
--- a/wasm-udf-inline/Cargo.toml
+++ b/examples/wasm-udf-inline/Cargo.toml
@@ -18,6 +18,12 @@
# Single-file WASM UDF example: the UDF *and* the client live in `src/main.rs`.
# `build.rs` (via the spark-connect-build helper) compiles that same file to a
# wasm32 module and embeds it.
+
+# Standalone crate with its own workspace root: it lives under `examples/` but
is
+# excluded from the root workspace because it targets `wasm32-unknown-unknown`.
+# (Nothing in the root workspace depends on it, so its own `[workspace]` is
safe.)
+[workspace]
+
[package]
name = "wasm-udf-inline"
version = "0.0.0"
@@ -25,16 +31,16 @@ edition = "2021"
publish = false
[dependencies]
-spark-connect-macros = { package = "apache-spark-connect-macros", path =
"../crates/spark-connect-macros", version = "4.2.0" }
+spark-connect-macros = { package = "apache-spark-connect-macros", path =
"../../crates/spark-connect-macros", version = "4.2.0" }
# Host-only: the client + the macro-generated constructors (which need the
# `wasm-udf` feature). Never compiled for wasm32.
[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
-spark-connect = { package = "apache-spark-connect", path =
"../crates/spark-connect", version = "4.2.0", features = ["wasm-udf"] }
+spark-connect = { package = "apache-spark-connect", path =
"../../crates/spark-connect", version = "4.2.0", features = ["wasm-udf"] }
[build-dependencies]
# The build helper that compiles + embeds the wasm module.
-spark-connect-build = { package = "apache-spark-connect-build", path =
"../crates/spark-connect-build", version = "4.2.0" }
+spark-connect-build = { package = "apache-spark-connect-build", path =
"../../crates/spark-connect-build", version = "4.2.0" }
# Forces the proc-macro to be built before build.rs runs, so the helper can
find
# its dylib to feed the wasm `rustc` invocation.
-spark-connect-macros = { package = "apache-spark-connect-macros", path =
"../crates/spark-connect-macros", version = "4.2.0" }
+spark-connect-macros = { package = "apache-spark-connect-macros", path =
"../../crates/spark-connect-macros", version = "4.2.0" }
diff --git a/wasm-udf-inline/build.rs b/examples/wasm-udf-inline/build.rs
similarity index 100%
rename from wasm-udf-inline/build.rs
rename to examples/wasm-udf-inline/build.rs
diff --git a/wasm-udf-inline/src/main.rs b/examples/wasm-udf-inline/src/main.rs
similarity index 100%
rename from wasm-udf-inline/src/main.rs
rename to examples/wasm-udf-inline/src/main.rs
diff --git a/wasm-udfs/Cargo.toml b/examples/wasm-udfs/Cargo.toml
similarity index 71%
rename from wasm-udfs/Cargo.toml
rename to examples/wasm-udfs/Cargo.toml
index 67ff354..73e1e3f 100644
--- a/wasm-udfs/Cargo.toml
+++ b/examples/wasm-udfs/Cargo.toml
@@ -18,6 +18,10 @@
# A reusable UDF crate: `#[spark_wasm_udf]` functions whose `<name>_udf()`
# constructors other crates can call. `build.rs` compiles this crate's own
# functions to wasm32 and embeds the module.
+#
+# No own `[workspace]` table: this crate is a path-dependency of the `examples`
+# member, so it must resolve within the root workspace (which `exclude`s it
from
+# membership). It builds standalone via `--manifest-path` because it's
excluded.
[package]
name = "wasm-udfs"
version = "0.0.0"
@@ -25,13 +29,13 @@ edition = "2021"
publish = false
[dependencies]
-spark-connect-macros = { package = "apache-spark-connect-macros", path =
"../crates/spark-connect-macros", version = "4.2.0" }
+spark-connect-macros = { package = "apache-spark-connect-macros", path =
"../../crates/spark-connect-macros", version = "4.2.0" }
# Host-only: the macro-generated constructors (which need the `wasm-udf`
# feature). Never compiled for wasm32.
[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
-spark-connect = { package = "apache-spark-connect", path =
"../crates/spark-connect", version = "4.2.0", features = ["wasm-udf"] }
+spark-connect = { package = "apache-spark-connect", path =
"../../crates/spark-connect", version = "4.2.0", features = ["wasm-udf"] }
[build-dependencies]
-spark-connect-build = { package = "apache-spark-connect-build", path =
"../crates/spark-connect-build", version = "4.2.0" }
-spark-connect-macros = { package = "apache-spark-connect-macros", path =
"../crates/spark-connect-macros", version = "4.2.0" }
+spark-connect-build = { package = "apache-spark-connect-build", path =
"../../crates/spark-connect-build", version = "4.2.0" }
+spark-connect-macros = { package = "apache-spark-connect-macros", path =
"../../crates/spark-connect-macros", version = "4.2.0" }
diff --git a/wasm-udfs/build.rs b/examples/wasm-udfs/build.rs
similarity index 100%
rename from wasm-udfs/build.rs
rename to examples/wasm-udfs/build.rs
diff --git a/wasm-udfs/src/lib.rs b/examples/wasm-udfs/src/lib.rs
similarity index 100%
rename from wasm-udfs/src/lib.rs
rename to examples/wasm-udfs/src/lib.rs
diff --git a/pre-commit.sh b/pre-commit.sh
deleted file mode 100644
index f82390e..0000000
--- a/pre-commit.sh
+++ /dev/null
@@ -1,86 +0,0 @@
-#!/bin/bash
-
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-# This file is git pre-commit hook.
-#
-# Soft link it as git hook under top dir of apache arrow git repository:
-# $ ln -s ../../pre-commit.sh .git/hooks/pre-commit
-#
-# This file be run directly:
-# $ ./pre-commit.sh
-
-function RED() {
- echo "\033[0;31m$@\033[0m"
-}
-
-function GREEN() {
- echo "\033[0;32m$@\033[0m"
-}
-
-function BYELLOW() {
- echo "\033[1;33m$@\033[0m"
-}
-
-# env GIT_DIR is set by git when run a pre-commit hook.
-if [ -z "${GIT_DIR}" ]; then
- GIT_DIR=$(git rev-parse --show-toplevel)
-fi
-
-cd ${GIT_DIR}
-
-NUM_CHANGES=$(git diff --cached --name-only . |
- grep -e ".*/*.rs$" |
- awk '{print $1}' |
- wc -l)
-
-if [ ${NUM_CHANGES} -eq 0 ]; then
- echo -e "$(GREEN INFO): no staged changes in *.rs, $(GREEN skip cargo
fmt/clippy)"
- exit 0
-fi
-
-# 1. cargo clippy
-
-echo -e "$(GREEN INFO): cargo clippy ..."
-
-# Cargo clippy always return exit code 0, and `tee` doesn't work.
-# So let's just run cargo clippy.
-cargo clippy
-echo -e "$(GREEN INFO): cargo clippy done"
-
-# 2. cargo fmt: format with nightly and stable.
-
-CHANGED_BY_CARGO_FMT=false
-echo -e "$(GREEN INFO): cargo fmt with nightly and stable ..."
-
-for version in nightly stable; do
- CMD="cargo +${version} fmt"
- ${CMD} --all -q -- --check 2>/dev/null
- if [ $? -ne 0 ]; then
- ${CMD} --all
- echo -e "$(BYELLOW WARN): ${CMD} changed some files"
- CHANGED_BY_CARGO_FMT=true
- fi
-done
-
-if ${CHANGED_BY_CARGO_FMT}; then
- echo -e "$(RED FAIL): git commit $(RED ABORTED), please have a look and
run git add/commit again"
- exit 1
-fi
-
-exit 0
diff --git a/python/pyspark_wasm_udf/pack.py b/python/pyspark_wasm_udf/pack.py
index 9b3c305..a38a7cb 100644
--- a/python/pyspark_wasm_udf/pack.py
+++ b/python/pyspark_wasm_udf/pack.py
@@ -35,10 +35,10 @@ generically with
:func:`pyspark.sql.types._parse_datatype_json_value`, so any
Spark type is supported without a token table.
The command mirrors ``pyspark.sql.connect.expressions.PythonUDF``:
-``cloudpickle.dumps((func, output_type))``. We use the standalone PyPI
-``cloudpickle`` (byte-for-byte the same as ``pyspark.cloudpickle``) and
register
-:mod:`pyspark_wasm_udf` for pickling **by value**, so executors need not have
-this package installed.
+``cloudpickle.dumps((func, output_type))``. We use the bundled
+``pyspark.cloudpickle`` (the same cloudpickle the reference client ships) and
+register :mod:`pyspark_wasm_udf` for pickling **by value**, so executors need
not
+have this package installed.
"""
import base64
diff --git a/wasm_udf_example.txt b/wasm_udf_example.txt
deleted file mode 100644
index e3b2172..0000000
--- a/wasm_udf_example.txt
+++ /dev/null
@@ -1,175 +0,0 @@
-================================================================================
-Rust WASM UDF on Spark — end-user example
-================================================================================
-
-Run a Rust user-defined function on Spark, distributed on the executors, with
no
-server-side plugin. You write plain Rust functions; the toolchain compiles them
-to WebAssembly, embeds the module, and ships it inside a normal Spark
PythonUDF.
-Executors need only the `wasmtime` Python package.
-
-An end user writes THREE small files.
-
-
---------------------------------------------------------------------------------
-1. src/main.rs — write plain Rust functions and use them
---------------------------------------------------------------------------------
-
- use spark_connect_macros::spark_wasm_udf;
-
- // Your UDFs — just plain Rust functions. Annotate the module; that's it.
- #[spark_wasm_udf]
- mod udfs {
- pub fn add_one(x: i64) -> i64 {
- x + 1
- }
-
- pub fn shout(s: String) -> String {
- format!("{}!", s.to_uppercase())
- }
- }
-
- #[cfg(not(target_arch = "wasm32"))]
- fn main() -> Result<(), Box<dyn std::error::Error>> {
- use spark_connect::functions::col;
- use spark_connect::{SparkSession, SparkSessionBuilder};
-
- let spark: SparkSession = SparkSessionBuilder::default()
- .remote("sc://127.0.0.1:15002/")
- .get_or_create()?;
-
- // Pass columns straight in — one per argument. The macro generated
- // udf::add_one / udf::shout: signature inferred, wasm module embedded.
- spark
- .range(5)?
- .select(vec![
- col("id"),
- udf::add_one(col("id"))?.alias("id_plus_one"),
-
udf::shout(col("id").cast(spark_connect::types::DataType::String {
- collation: "UTF8_BINARY".to_string(),
- }))?
- .alias("shouted"),
- ])
- .show(20)?;
-
- Ok(())
- }
-
-
---------------------------------------------------------------------------------
-2. build.rs — one line
---------------------------------------------------------------------------------
-
- fn main() {
- spark_connect_build::embed_wasm_udf("src/main.rs");
- }
-
-
---------------------------------------------------------------------------------
-3. Cargo.toml — dependencies
---------------------------------------------------------------------------------
-
- [dependencies]
- spark-connect-macros = { package = "apache-spark-connect-macros", version
= "4.2" }
-
- # client is host-only; not pulled into the wasm build
- [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
- spark-connect = { package = "apache-spark-connect", version = "4.2",
features = ["wasm-udf"] }
-
- [build-dependencies]
- spark-connect-build = { package = "apache-spark-connect-build", version
= "4.2" }
- spark-connect-macros = { package = "apache-spark-connect-macros", version
= "4.2" }
-
-
---------------------------------------------------------------------------------
-Run it
---------------------------------------------------------------------------------
-
- rustup target add wasm32-unknown-unknown # once
- export SPARK_CONNECT_WASM_PACKER_PATH=$PWD/python # so the client finds
the packer
- cargo run
-
-Output (against a Spark 4.2.0 Connect server on :15002, wasmtime on executors):
-
- +---+-----------+-------+
- | id|id_plus_one|shouted|
- +---+-----------+-------+
- | 0| 1| 0!|
- | 1| 2| 1!|
- | 2| 3| 2!|
- ...
- +---+-----------+-------+
-
-
---------------------------------------------------------------------------------
-The end-user mental model
---------------------------------------------------------------------------------
-
- They write They call They never touch
- ----------------------- ------------------------
------------------------
- fn add_one(x: i64) -> i64 udf::add_one(col("id"))? WASM bytes, the
ABI,
- a .wasm file, the
Spark
- type of the UDF
-
-The macro infers the Spark signature from the Rust types ((Long) -> Long),
-build.rs compiles the SAME source to WASM and embeds it, and the client ships
it
-inside a normal Spark PythonUDF. Arity is checked at compile time: add_one
takes
-one column; a two-arg UDF would be udf::concat(col("a"), col("b"))?.
-
-
---------------------------------------------------------------------------------
-Supported argument / result types (inferred; nest arbitrarily)
---------------------------------------------------------------------------------
-
- Rust Spark SQL type
- ---------- ---------------------
- i32 IntegerType
- i64 LongType
- f32 FloatType
- f64 DoubleType
- bool BooleanType
- String StringType
- Vec<u8> BinaryType
- Vec<T> ArrayType (of T)
- Option<T> nullable T
-
-e.g. Vec<Option<String>> -> ArrayType(StringType, nullable).
-
-
---------------------------------------------------------------------------------
-Two other shapes
---------------------------------------------------------------------------------
-
-- UDFs as a reusable crate, called from any client:
- wasm_udfs::udf::add_one(col("id"))?
- (see wasm-udfs/ + examples/src/wasm_udf_macro.rs)
-
-- Advanced config via the builder accessor:
- udf::add_one_udf().as_nondeterministic().call(vec![col("id")])?
-
-
---------------------------------------------------------------------------------
-Caveat (see review finding #3 on PR #50)
---------------------------------------------------------------------------------
-
-The UDF currently reports Python "3.11" to Spark by default. If your executors
-run a different Python minor version, set it explicitly until the default is
-fixed to derive from the packer's Python:
-
- udf::add_one_udf().with_python_ver("3.10").call(vec![col("id")])?
-
-
---------------------------------------------------------------------------------
-Preconditions (only when using Rust UDFs; nothing here unless `wasm-udf` is on)
---------------------------------------------------------------------------------
-
-- Build machine: wasm32-unknown-unknown target; apache-spark-connect-macros +
- apache-spark-connect-build as (build-)dependencies.
-- Client (building the UDF command): a Python interpreter with `cloudpickle`
- and `pyspark`, plus the repo's python/ directory importable as
- `pyspark_wasm_udf` (SPARK_CONNECT_PYTHON / SPARK_CONNECT_WASM_PACKER_PATH, or
- UserDefinedFunction::with_packer).
-- Executors: the `wasmtime` Python package installed.
-- Spark: 4.2.0+.
-
-The single-file example above lives in the repo at wasm-udf-inline/.
-================================================================================
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
