Visitar URL original
feat: Bring Your Own Spark - SparkApplication by aniketpalu · Pull Request #6550 · feast-dev/feast · GitHub
Skip to content

feat: Bring Your Own Spark - SparkApplication #6550

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ntkathole merged 16 commits into from
Jul 17, 2026
Merged

feat: Bring Your Own Spark - SparkApplication #6550

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
462ad7b
feat: Add SparkApplicationComputeEngine for Kubernetes-native batch m…
aniketpalu Jun 24, 2026
9d4a125
feat: Add per-FV result reporting and clean up Dockerfile
aniketpalu Jul 3, 2026
0193c52
Minor lint & formatting change
aniketpalu Jul 3, 2026
79eb653
Merge branch 'master' into feat/spark-application-compute-engine
aniketpalu Jul 3, 2026
a71de0c
Merge branch 'master' into feat/spark-application-compute-engine
aniketpalu Jul 8, 2026
4d49be9
feat: Switch to ConfigMap, remove registry_address, reject file-based…
aniketpalu Jul 9, 2026
7a27217
Merge branch 'master' into feat/spark-application-compute-engine
aniketpalu Jul 14, 2026
9879506
Minor formatting
aniketpalu Jul 15, 2026
d3fc4c8
fix: address PR #6550 review — retry, validation, per-FV status
aniketpalu Jul 16, 2026
ba73932
Merge branch 'master' into feat/spark-application-compute-engine
aniketpalu Jul 16, 2026
a38a56d
fix: isolate batch materialization to supports_batch engines
aniketpalu Jul 16, 2026
0f348a8
fix: address ntkathole review — dates dataclass, jobs check, UNKNOWN,…
aniketpalu Jul 16, 2026
cef6fa6
fix: lazy-init K8s client for spark_application engine
aniketpalu Jul 16, 2026
5c0b9fb
chore: Refresh pixi.lock after pyproject.toml dependency changes
aniketpalu Jul 16, 2026
396151e
fix: skip duplicate apply_materialization for SparkApplication
aniketpalu Jul 17, 2026
cbd4051
fix: harden supports_batch check against missing batch_engine
aniketpalu Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
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>
  • Loading branch information
@aniketpalu
aniketpalu committed Jul 16, 2026
commit a38a56dda4937564cb8fd13abb34be86cc695467
258 changes: 194 additions & 64 deletions sdk/python/feast/feature_store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -527,6 +527,51 @@ def _submit_and_process_materialization_jobs(
if first_error:
raise first_error

def _materialize_fvs_batch(
self,
provider,
fv_with_dates: list,
end_date: datetime,
tqdm_builder,
disable_event_timestamp: bool = False,
) -> None:
"""Batch path: collect all FVs, submit to engine in one call.

Only used when ``provider.batch_engine.supports_batch`` is True.
"""
from feast.infra.common.materialization_job import MaterializationTask

tasks: list = []
regular_fvs: list = []
previous_states: dict = {}
fv_start_dates: dict = {"__end_date__": end_date}

for feature_view, fv_start in fv_with_dates:
self._transition_fv_to_materializing(
feature_view, regular_fvs, previous_states
)
regular_fvs.append(feature_view)
fv_start_dates[feature_view.name] = fv_start
tasks.append(
MaterializationTask(
project=self.project,
feature_view=feature_view,
start_time=fv_start,
end_time=end_date,
tqdm_builder=tqdm_builder,
disable_event_timestamp=disable_event_timestamp,
)
)

if tasks:
self._submit_and_process_materialization_jobs(
provider,
tasks,
regular_fvs,
previous_states,
fv_start_dates,
)

@property
def openlineage_emitter(self) -> Optional[Any]:
"""Gets the OpenLineage emitter of this feature store."""
Expand Down Expand Up @@ -2352,20 +2397,14 @@ def materialize_incremental(

_mat_start = time.monotonic()
try:
from feast.infra.common.materialization_job import (
MaterializationTask,
)

provider = self._get_provider()
end_date_tz = utils.make_tzaware(end_date) or _utc_now()

def tqdm_builder(length):
return tqdm(total=length, ncols=100)

tasks: list = []
regular_fvs: list = []
previous_states: dict = {}
fv_start_dates: dict = {"__end_date__": end_date_tz}
# (feature_view, start_date) — start_date is always set before append
regular_fvs_with_dates: list[tuple[Any, datetime]] = []

for feature_view in feature_views_to_materialize:
if isinstance(feature_view, OnDemandFeatureView):
Expand Down Expand Up @@ -2395,55 +2434,103 @@ def tqdm_builder(length):
)
continue

fv_start_date = feature_view.most_recent_end_time
if fv_start_date is None:
start_date = feature_view.most_recent_end_time
if start_date is None:
if feature_view.ttl is None:
raise Exception(
f"No start time found for feature view {feature_view.name}. materialize_incremental() requires"
f" either a ttl to be set or for materialize() to have been run at least once."
)
elif feature_view.ttl.total_seconds() > 0:
fv_start_date = _utc_now() - feature_view.ttl
start_date = _utc_now() - feature_view.ttl
else:
# TODO(felixwang9817): Find the earliest timestamp for this specific feature
# view from the offline store, and set the start date to that timestamp.
print(
f"Since the ttl is 0 for feature view {Style.BRIGHT + Fore.GREEN}{feature_view.name}{Style.RESET_ALL}, "
"the start date will be set to 1 year before the current time."
)
fv_start_date = _utc_now() - timedelta(weeks=52)

fv_start_date = utils.make_tzaware(fv_start_date)
fv_start_dates[feature_view.name] = fv_start_date
start_date = _utc_now() - timedelta(weeks=52)

start_date = utils.make_tzaware(start_date)
print(
f"{Style.BRIGHT + Fore.GREEN}{feature_view.name}{Style.RESET_ALL}"
f" from {Style.BRIGHT + Fore.GREEN}{utils.make_tzaware(fv_start_date.replace(microsecond=0))}{Style.RESET_ALL}"
f" from {Style.BRIGHT + Fore.GREEN}{utils.make_tzaware(start_date.replace(microsecond=0))}{Style.RESET_ALL}"
f" to {Style.BRIGHT + Fore.GREEN}{utils.make_tzaware(end_date.replace(microsecond=0))}{Style.RESET_ALL}:"
)

self._transition_fv_to_materializing(
feature_view, regular_fvs, previous_states
)
regular_fvs.append(feature_view)
tasks.append(
MaterializationTask(
project=self.project,
feature_view=feature_view,
start_time=fv_start_date,
end_time=end_date_tz,
tqdm_builder=tqdm_builder,
)
)
regular_fvs_with_dates.append((feature_view, start_date))

if tasks:
self._submit_and_process_materialization_jobs(
# batch_engine is on PassthroughProvider (concrete); same access as
# _submit_and_process_materialization_jobs via untyped provider.
if getattr(provider, "batch_engine").supports_batch:

@ntkathole ntkathole Jul 17, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fragile

batch_engine = getattr(provider, "batch_engine", None)
if batch_engine and getattr(batch_engine, "supports_batch", False):

@aniketpalu aniketpalu Jul 17, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

self._materialize_fvs_batch(
provider,
tasks,
regular_fvs,
previous_states,
fv_start_dates,
regular_fvs_with_dates,
end_date_tz,
tqdm_builder,
)
else:
for feature_view, start_date in regular_fvs_with_dates:
# Transition state to MATERIALIZING before starting.
# Only enforce when the state machine is active (not STATE_UNSPECIFIED).
previous_state = getattr(feature_view, "state", None)
if (
hasattr(feature_view, "state")
and feature_view.state != FeatureViewState.STATE_UNSPECIFIED
):
if not feature_view.state.can_transition_to(
FeatureViewState.MATERIALIZING
):
raise ValueError(
f"FeatureView {feature_view.name} cannot transition "
f"from {feature_view.state.name} to MATERIALIZING."
)
feature_view.state = FeatureViewState.MATERIALIZING
self.registry.apply_feature_view(
feature_view, self.project, commit=True
)

fv_start = time.monotonic()
fv_success = True
try:
provider.materialize_single_feature_view(
config=self.config,
feature_view=feature_view,
start_date=start_date,
end_date=end_date_tz,
registry=self.registry,
project=self.project,
tqdm_builder=tqdm_builder,
)
except Exception:
fv_success = False
# Roll back state to previous value on failure.
if (
hasattr(feature_view, "state")
and previous_state is not None
and previous_state != FeatureViewState.STATE_UNSPECIFIED
):
feature_view.state = previous_state
self.registry.apply_feature_view(
feature_view, self.project, commit=True
)
raise
finally:
_tracker = _get_track_materialization()
if _tracker is not None:
_tracker(
feature_view.name,
fv_success,
time.monotonic() - fv_start,
)

self.registry.apply_materialization(
feature_view,
self.project,
start_date,
end_date_tz,
)

materialized_fv_names = [
fv.name
Expand Down Expand Up @@ -2535,21 +2622,14 @@ def materialize(

_mat_start = time.monotonic()
try:
from feast.infra.common.materialization_job import (
MaterializationTask,
)

provider = self._get_provider()
start_date = utils.make_tzaware(start_date)
end_date = utils.make_tzaware(end_date)

def tqdm_builder(length):
return tqdm(total=length, ncols=100)

tasks: list = []
regular_fvs: list = []
previous_states: dict = {}
fv_start_dates: dict = {"__end_date__": end_date}
regular_fvs_with_dates: list[tuple[Any, datetime]] = []

for feature_view in feature_views_to_materialize:
if isinstance(feature_view, OnDemandFeatureView):
Expand All @@ -2569,30 +2649,80 @@ def tqdm_builder(length):
f"{Style.BRIGHT + Fore.GREEN}{feature_view.name}{Style.RESET_ALL}:"
)

self._transition_fv_to_materializing(
feature_view, regular_fvs, previous_states
)
regular_fvs.append(feature_view)
fv_start_dates[feature_view.name] = start_date
tasks.append(
MaterializationTask(
project=self.project,
feature_view=feature_view,
start_time=start_date,
end_time=end_date,
tqdm_builder=tqdm_builder,
disable_event_timestamp=disable_event_timestamp,
)
)
regular_fvs_with_dates.append((feature_view, start_date))

if tasks:
self._submit_and_process_materialization_jobs(
# batch_engine is on PassthroughProvider (concrete); same access as
# _submit_and_process_materialization_jobs via untyped provider.
if getattr(provider, "batch_engine").supports_batch:
self._materialize_fvs_batch(
provider,
tasks,
regular_fvs,
previous_states,
fv_start_dates,
regular_fvs_with_dates,
end_date,
tqdm_builder,
disable_event_timestamp=disable_event_timestamp,
)
else:
for feature_view, fv_start in regular_fvs_with_dates:
# Transition state to MATERIALIZING before starting.
# Only enforce when the state machine is active (not STATE_UNSPECIFIED).
previous_state = getattr(feature_view, "state", None)
if (
hasattr(feature_view, "state")
and feature_view.state != FeatureViewState.STATE_UNSPECIFIED
):
if not feature_view.state.can_transition_to(
FeatureViewState.MATERIALIZING
):
raise ValueError(
f"FeatureView {feature_view.name} cannot transition "
f"from {feature_view.state.name} to MATERIALIZING."
)
feature_view.state = FeatureViewState.MATERIALIZING
self.registry.apply_feature_view(
feature_view, self.project, commit=True
)

fv_start_time = time.monotonic()
fv_success = True
try:
provider.materialize_single_feature_view(
config=self.config,
feature_view=feature_view,
start_date=fv_start,
end_date=end_date,
registry=self.registry,
project=self.project,
tqdm_builder=tqdm_builder,
disable_event_timestamp=disable_event_timestamp,
)
except Exception:
fv_success = False
# Roll back state to previous value on failure.
if (
hasattr(feature_view, "state")
and previous_state is not None
and previous_state != FeatureViewState.STATE_UNSPECIFIED
):
feature_view.state = previous_state
self.registry.apply_feature_view(
feature_view, self.project, commit=True
)
raise
finally:
_tracker = _get_track_materialization()
if _tracker is not None:
_tracker(
feature_view.name,
fv_success,
time.monotonic() - fv_start_time,
)

self.registry.apply_materialization(
feature_view,
self.project,
fv_start,
end_date,
)

materialized_fv_names = [
fv.name
Expand Down
10 changes: 10 additions & 0 deletions sdk/python/feast/infra/compute_engines/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,16 @@ def teardown_infra(
"""
pass

@property
def supports_batch(self) -> bool:
"""Whether this engine can accept all tasks in a single materialize() call.

When True, feature_store.py collects all FV tasks upfront and submits
them in one batch. When False (default), the standard per-FV loop
through ``provider.materialize_single_feature_view()`` is used.
"""
return False

def materialize(
self,
registry: BaseRegistry,
Expand Down
4 changes: 4 additions & 0 deletions sdk/python/feast/infra/compute_engines/spark_application/compute.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,10 @@ def __init__(
self.custom_api = client.CustomObjectsApi(self.k8s_client)
self._server_id = uuid.uuid4().hex[:8]

@property
def supports_batch(self) -> bool:
return True

def update(
self,
project: str,
Expand Down
Loading