Skip to content

Commit dcd496f

Browse files
aniketpaluaniketpalu
authored
feat: Bring Your Own Spark - SparkApplication (#6550)
* feat: Add SparkApplicationComputeEngine for Kubernetes-native batch materialization Adds a new batch compute engine that submits materialization jobs as SparkApplication CRDs via the Kubeflow Spark Operator. One 'feast materialize' call creates one SparkApplication pod that processes all feature views using distributed Spark, rather than running in-process on the Feast server. Key changes: - Refactor materialize()/materialize_incremental() to pass all tasks to the engine in a single batch call instead of looping per feature view. Existing engines are unaffected (base class loops tasks internally via _materialize_one). - Add public get_provider() method on FeatureStore. - New spark_application engine: config, compute, job, driver script, Dockerfile. - 12 unit tests covering config, validation, CR structure, state mapping, timeout, cleanup, and job naming. Signed-off-by: Aniket Paluskar <apaluska@redhat.com> * feat: Add per-FV result reporting and clean up Dockerfile - Pod calls apply_materialization via gRPC after each successful FV, setting state to AVAILABLE_ONLINE. Server reads FV state post-completion to determine per-FV success/failure in batched SparkApplication runs. - registry_address is now mandatory (simplified from complex path heuristic). - Dockerfile rewritten to install feast from source (matches K8s engine pattern). - Unit tests updated: 15/15 pass (3 new tests for _build_per_fv_jobs). - E2E validated: 5 FVs x 9600 rows, 5 executors, all AVAILABLE_ONLINE. Signed-off-by: Aniket Paluskar <apaluska@redhat.com> * Minor lint & formatting change Signed-off-by: Aniket Paluskar <apaluska@redhat.com> * feat: Switch to ConfigMap, remove registry_address, reject file-based stores - Config delivery: Secret → ConfigMap. Operator's ClusterRole already has full ConfigMap CRUD — avoids widening RBAC for Secrets in ODH. Matches KubernetesComputeEngine pattern. - Removed registry_address config field. Pod inherits server's registry config (SQL, Snowflake) directly and writes apply_materialization() to the same database. Eliminates TLS certificate mounting complexity. - Reject file-based offline stores (dask, file, duckdb) and registries (file) at __init__(), same as existing sqlite/faiss online store rejection. SparkApplication pod has ephemeral filesystem. - Dockerfile: added PYTHONPATH/SPARK_HOME for PySpark, added pymysql. - 20/20 unit tests pass. Signed-off-by: Aniket Paluskar <apaluska@redhat.com> * Minor formatting Signed-off-by: Aniket Paluskar <apaluska@redhat.com> * fix: address PR #6550 review — retry, validation, per-FV status - Remove redundant get_provider(); callers use .provider property - Replace SparkSession monkey-patch with per-thread session binding - Smart retry: only 5xx/429; fail fast on 401/403 with RBAC hint - Retry ConfigMap + SparkApplication creation (transient K8s errors) - Validate env entries: require name + value or valueFrom (K8s EnvVar) - Independent per-FV job status via CompletedMaterializationJob - Exit 1 on any FV failure so SparkApp CR reflects partial failure - Cleanup logs include kubectl delete command for manual recovery - Lint: pragma allowlist on test fixture URL Signed-off-by: Aniket Paluskar <apaluska@redhat.com> * fix: isolate batch materialization to supports_batch engines Restore master's per-FV materialize path for engines that do not support batching; keep SparkApplication on the existing batch path via ComputeEngine.supports_batch. Signed-off-by: Aniket Paluskar <apaluska@redhat.com> * fix: address ntkathole review — dates dataclass, jobs check, UNKNOWN, cleanup Replace __end_date__ sentinel with _MaterializationDateRange; fail fast if engine job count mismatches; map UNKNOWN SparkApp state to WAITING; always cleanup ConfigMap/CR after wait. Signed-off-by: Aniket Paluskar <apaluska@redhat.com> * fix: lazy-init K8s client for spark_application engine Defer kubeconfig load until materialize/cleanup so feast apply can construct the engine without a cluster. Signed-off-by: Aniket Paluskar <apaluska@redhat.com> * chore: Refresh pixi.lock after pyproject.toml dependency changes Signed-off-by: Aniket Paluskar <apaluska@redhat.com> * fix: skip duplicate apply_materialization for SparkApplication Pod already writes watermarks via applies_materialization; server batch path skips the second call to avoid duplicate intervals. Signed-off-by: Aniket Paluskar <apaluska@redhat.com> * fix: harden supports_batch check against missing batch_engine Use null-safe getattr so materialize falls back to the per-FV path when batch_engine or supports_batch is absent. Signed-off-by: Aniket Paluskar <apaluska@redhat.com> --------- Signed-off-by: Aniket Paluskar <apaluska@redhat.com>
1 parent 14e0a83 commit dcd496f

12 files changed

Lines changed: 1822 additions & 168 deletions

File tree

‎pixi.lock‎

Lines changed: 55 additions & 54 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎sdk/python/feast/feature_store.py‎

Lines changed: 313 additions & 114 deletions
Large diffs are not rendered by default.

‎sdk/python/feast/infra/compute_engines/base.py‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,21 @@ def teardown_infra(
8585
"""
8686
pass
8787

88+
@property
89+
def supports_batch(self) -> bool:
90+
"""Whether this engine can accept all tasks in a single materialize() call.
91+
92+
When True, feature_store.py collects all FV tasks upfront and submits
93+
them in one batch. When False (default), the standard per-FV loop
94+
through ``provider.materialize_single_feature_view()`` is used.
95+
"""
96+
return False
97+
98+
@property
99+
def applies_materialization(self) -> bool:
100+
"""If True, the engine already wrote watermarks/state (e.g. driver pod)."""
101+
return False
102+
88103
def materialize(
89104
self,
90105
registry: BaseRegistry,

‎sdk/python/feast/infra/compute_engines/spark_application/.dockerignore‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
feast/.git
2+
feast/__pycache__
3+
feast/**/__pycache__
4+
feast/.mypy_cache
5+
feast/tests
6+
feast/.pixi
7+
**/*.pyc
8+
**/.pytest_cache

‎sdk/python/feast/infra/compute_engines/spark_application/Dockerfile‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
FROM apache/spark:4.0.1
2+
3+
USER root
4+
5+
RUN apt-get update && apt-get install --no-install-suggests --no-install-recommends --yes git && rm -rf /var/lib/apt/lists/*
6+
7+
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
8+
9+
ENV PYTHONPATH="/opt/spark/python:/opt/spark/python/lib/py4j-0.10.9.9-src.zip"
10+
ENV SPARK_HOME="/opt/spark"
11+
12+
WORKDIR /app
13+
14+
COPY sdk/python/feast/infra/compute_engines/spark_application/main.py /opt/feast/main.py
15+
16+
# Copy necessary parts of the Feast codebase
17+
COPY sdk/python sdk/python
18+
COPY protos protos
19+
COPY pyproject.toml pyproject.toml
20+
COPY README.md README.md
21+
22+
# setuptools_scm needs .git to infer the version.
23+
# https://github.com/pypa/setuptools_scm#usage-from-docker
24+
RUN --mount=source=.git,target=.git,type=bind uv pip install --system --no-cache-dir '.[redis,grpcio,k8s]' pymysql
25+
26+
# Hadoop AWS JARs for S3A filesystem support + AWS SDK v2 bundle
27+
RUN curl -sL https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.4.1/hadoop-aws-3.4.1.jar \
28+
-o /opt/spark/jars/hadoop-aws-3.4.1.jar && \
29+
curl -sL https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.12.782/aws-java-sdk-bundle-1.12.782.jar \
30+
-o /opt/spark/jars/aws-java-sdk-bundle-1.12.782.jar && \
31+
curl -sL https://repo1.maven.org/maven2/software/amazon/awssdk/bundle/2.28.4/bundle-2.28.4.jar \
32+
-o /opt/spark/jars/bundle-2.28.4.jar
33+
34+
USER spark

‎sdk/python/feast/infra/compute_engines/spark_application/__init__.py‎

Whitespace-only changes.

0 commit comments

Comments
 (0)