fix(spark): Pre-create S3A event log dir before SparkContext init to prevent silent materialize failure

abhijeet-dhumal · abhijeet-dhumal · commit 2139cff89a66 · 2026-04-22T21:54:47.000+05:30
Spark's EventLogFileWriter.requireLogBaseDirAsDirectory() is called
inside SparkContext.__init__. When spark.eventLog.dir points to an S3A
path that doesn't exist yet (S3 has no real directories), SparkContext
fails to initialise &mdash; silently from Feast's perspective because
_materialize_one() catches the exception and returns an ERROR job.

Add _ensure_s3a_event_log_dir() to utils.py: before building the
SparkSession, check if the S3A prefix exists and write a zero-byte
placeholder if it doesn't. Uses boto3 (already a Feast dep via S3 offline
store). Non-fatal: logs a warning and lets Spark surface its own error
if the write fails.

Signed-off-by: abhijeet-dhumal &lt;abhijeetdhumal652@gmail.com&gt;
diff --git a/sdk/python/feast/infra/compute_engines/spark/utils.py b/sdk/python/feast/infra/compute_engines/spark/utils.py
@@ -1,3 +1,5 @@
+import logging
+import os
 from typing import Dict, Iterable, Literal, Optional
 
 import pandas as pd
@@ -9,6 +11,68 @@
 from feast.infra.common.serde import SerializedArtifacts
 from feast.utils import _convert_arrow_to_proto, _run_pyarrow_field_mapping
 
+logger = logging.getLogger(__name__)
+
+
+def _ensure_s3a_event_log_dir(spark_config: Dict[str, str]) -> None:
+    """Pre-create the S3A event log prefix before SparkContext initialisation.
+
+    Spark's EventLogFileWriter.requireLogBaseDirAsDirectory() is called inside
+    SparkContext.__init__ and crashes if the S3A path doesn't exist yet (S3 has no
+    real directories, so an empty prefix returns a 404). This function writes a
+    zero-byte placeholder so the prefix exists before SparkContext is built.
+
+    This is only attempted when:
+      - spark.eventLog.enabled == "true"
+      - spark.eventLog.dir starts with "s3a://"
+    Failures are non-fatal: Spark will surface its own error if the dir is still missing.
+    """
+    if spark_config.get("spark.eventLog.enabled", "false").lower() != "true":
+        return
+    event_dir = spark_config.get("spark.eventLog.dir", "")
+    if not event_dir.startswith("s3a://"):
+        return
+
+    path = event_dir[len("s3a://") :]
+    bucket, _, prefix = path.partition("/")
+    prefix = prefix.rstrip("/") + "/"
+    placeholder_key = prefix + ".keep"
+
+    endpoint = spark_config.get(
+        "spark.hadoop.fs.s3a.endpoint",
+        os.environ.get("FEAST_S3A_ENDPOINT", ""),
+    )
+    access_key = os.environ.get("AWS_ACCESS_KEY_ID", "")
+    secret_key = os.environ.get("AWS_SECRET_ACCESS_KEY", "")
+
+    try:
+        import boto3
+        from botocore.client import Config
+
+        s3 = boto3.client(
+            "s3",
+            endpoint_url=endpoint if endpoint else None,
+            aws_access_key_id=access_key or None,
+            aws_secret_access_key=secret_key or None,
+            config=Config(signature_version="s3v4"),
+        )
+        resp = s3.list_objects_v2(Bucket=bucket, Prefix=prefix, MaxKeys=1)
+        if resp.get("KeyCount", 0) == 0:
+            s3.put_object(Bucket=bucket, Key=placeholder_key, Body=b"")
+            logger.debug(
+                "Created S3A event log dir placeholder: s3a://%s/%s",
+                bucket,
+                placeholder_key,
+            )
+    except Exception as exc:
+        logger.warning(
+            "Could not pre-create S3A event log dir s3a://%s/%s &mdash; "
+            "SparkContext may fail if the path still doesn't exist: %s",
+            bucket,
+            prefix,
+            exc,
+        )
+
 
 def get_or_create_new_spark_session(
     spark_config: Optional[Dict[str, str]] = None,
@@ -17,6 +81,10 @@ def get_or_create_new_spark_session(
     if not spark_session:
         spark_builder = SparkSession.builder
         if spark_config:
+            # Spark's EventLogFileWriter.requireLogBaseDirAsDirectory() is called
+            # during SparkContext.__init__ and will crash if the S3A event log
+            # prefix doesn't exist yet. Ensure the prefix exists first.
+            _ensure_s3a_event_log_dir(spark_config)
             spark_builder = spark_builder.config(
                 conf=SparkConf().setAll([(k, v) for k, v in spark_config.items()])
             )
diff --git a/sdk/python/tests/component/spark/test_compute.py b/sdk/python/tests/component/spark/test_compute.py
@@ -1,6 +1,6 @@
 from datetime import timedelta
 from typing import cast
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 import pytest
 from pyspark.sql import DataFrame
@@ -15,6 +15,7 @@
 from feast.infra.common.retrieval_task import HistoricalRetrievalTask
 from feast.infra.compute_engines.spark.compute import SparkComputeEngine
 from feast.infra.compute_engines.spark.job import SparkDAGRetrievalJob
+from feast.infra.compute_engines.spark.utils import _ensure_s3a_event_log_dir
 from feast.infra.offline_stores.contrib.spark_offline_store.spark import (
     SparkOfflineStore,
 )
@@ -192,5 +193,76 @@ def tqdm_builder(length):
         spark_environment.teardown()
 
 
+# ---------------------------------------------------------------------------
+# Unit tests for _ensure_s3a_event_log_dir &mdash; no Spark dependency needed
+# ---------------------------------------------------------------------------
+
+
+def _base_conf(event_log_dir: str) -> dict:
+    return {
+        "spark.eventLog.enabled": "true",
+        "spark.eventLog.dir": event_log_dir,
+        "spark.hadoop.fs.s3a.endpoint": "http://minio:9000",
+    }
+
+
+@patch("feast.infra.compute_engines.spark.utils.boto3")
+def test_ensure_s3a_event_log_dir_creates_placeholder_when_empty(mock_boto3):
+    """S3A prefix doesn't exist &rarr; placeholder object is written."""
+    s3 = MagicMock()
+    mock_boto3.client.return_value = s3
+    s3.list_objects_v2.return_value = {"KeyCount": 0}
+
+    _ensure_s3a_event_log_dir(_base_conf("s3a://my-bucket/spark-events/"))
+
+    s3.list_objects_v2.assert_called_once_with(
+        Bucket="my-bucket", Prefix="spark-events/", MaxKeys=1
+    )
+    s3.put_object.assert_called_once_with(
+        Bucket="my-bucket", Key="spark-events/.keep", Body=b""
+    )
+
+
+@patch("feast.infra.compute_engines.spark.utils.boto3")
+def test_ensure_s3a_event_log_dir_skips_when_prefix_exists(mock_boto3):
+    """S3A prefix already has objects &rarr; no placeholder written."""
+    s3 = MagicMock()
+    mock_boto3.client.return_value = s3
+    s3.list_objects_v2.return_value = {"KeyCount": 3}
+
+    _ensure_s3a_event_log_dir(_base_conf("s3a://my-bucket/spark-events/"))
+
+    s3.put_object.assert_not_called()
+
+
+@patch("feast.infra.compute_engines.spark.utils.boto3")
+def test_ensure_s3a_event_log_dir_noop_when_event_log_disabled(mock_boto3):
+    """spark.eventLog.enabled != true &rarr; boto3 never called."""
+    _ensure_s3a_event_log_dir(
+        {"spark.eventLog.enabled": "false", "spark.eventLog.dir": "s3a://b/p/"}
+    )
+    mock_boto3.client.assert_not_called()
+
+
+@patch("feast.infra.compute_engines.spark.utils.boto3")
+def test_ensure_s3a_event_log_dir_noop_for_non_s3a_path(mock_boto3):
+    """Non-S3A paths (hdfs://, file://, etc.) are left untouched."""
+    _ensure_s3a_event_log_dir(
+        {"spark.eventLog.enabled": "true", "spark.eventLog.dir": "hdfs:///spark-logs"}
+    )
+    mock_boto3.client.assert_not_called()
+
+
+@patch("feast.infra.compute_engines.spark.utils.boto3")
+def test_ensure_s3a_event_log_dir_non_fatal_on_s3_error(mock_boto3):
+    """boto3 errors are swallowed &mdash; SparkContext will surface its own error."""
+    s3 = MagicMock()
+    mock_boto3.client.return_value = s3
+    s3.list_objects_v2.side_effect = Exception("connection refused")
+
+    # Must not raise
+    _ensure_s3a_event_log_dir(_base_conf("s3a://my-bucket/spark-events/"))
+
+
 if __name__ == "__main__":
     test_spark_compute_engine_get_historical_features()
