Merge pull request #28 from Neuw84/test/spark-geospatial

fix(aws): Make the in-image Iceberg jar path configurable

Author: Neuw84

.github/workflows/aws-platform-tests.yml

@@ -31,6 +31,16 @@ on:
         description: EMR release label
         type: string
         default: emr-spark-8.0.0
+      iceberg-jar-path:
+        description: >
+          In-image Iceberg jar for spark.jars. Release-dependent; leave empty to
+          omit spark.jars when Iceberg is already on the classpath.
+        type: string
+        default: /usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar
+      probe:
+        description: Run only the diagnostic job (reports the image layout)
+        type: boolean
+        default: false
       dry-run:
         description: Assume the role and stop (proves OIDC without spending)
         type: boolean
@@ -102,6 +112,8 @@ jobs:
         env:
           MODES: ${{ inputs.modes }}
           EMR_RELEASE_LABEL: ${{ inputs.release-label }}
+          ICEBERG_JAR_PATH: ${{ inputs.iceberg-jar-path }}
+          PROBE: ${{ inputs.probe }}
         run: uv run --with boto3 python tests/aws/run_emr_serverless.py
 
       - name: Tear down billable resources

tests/aws/emr_probe.py

@@ -0,0 +1,101 @@
+"""Diagnostic job: report what the EMR Serverless image actually provides.
+
+Submitted by run_emr_serverless.py when PROBE=1. Costs one short job run and
+answers the questions that otherwise turn into guesswork:
+
+  - where the Iceberg runtime jar lives, and what it is called
+  - whether Iceberg is already on the default classpath (no spark.jars needed)
+  - whether the S3 Tables catalog implementation is present
+  - which Spark and Iceberg versions the release label actually gives you
+
+Reads nothing and writes nothing. Deliberately does not create a SparkSession
+with Iceberg extensions, so it cannot fail for reasons unrelated to discovery.
+"""
+
+import glob
+import os
+import subprocess
+import sys
+
+CANDIDATE_DIRS = [
+    "/usr/share/aws/iceberg/lib",
+    "/usr/share/aws/iceberg",
+    "/usr/lib/spark/jars",
+    "/usr/share/aws/aws-java-sdk",
+    "/usr/share/aws/s3tables",
+]
+
+PATTERNS = ["*iceberg*", "*s3tables*", "*s3-tables*"]
+
+
+def section(title: str) -> None:
+    print(f"\nPROBE ===== {title} =====")
+
+
+def main() -> int:
+    section("python / spark")
+    print(f"PROBE python: {sys.version.split()[0]}")
+    try:
+        import pyspark
+        print(f"PROBE pyspark: {pyspark.__version__}")
+        print(f"PROBE SPARK_HOME: {os.environ.get('SPARK_HOME', '(unset)')}")
+    except Exception as e:  # noqa: BLE001
+        print(f"PROBE pyspark import failed: {e}")
+
+    section("candidate directories")
+    for d in CANDIDATE_DIRS:
+        if os.path.isdir(d):
+            entries = sorted(os.listdir(d))
+            print(f"PROBE dir {d}: {len(entries)} entries")
+            for name in entries[:40]:
+                print(f"PROBE     {name}")
+        else:
+            print(f"PROBE dir {d}: MISSING")
+
+    section("jar search")
+    for pattern in PATTERNS:
+        hits = []
+        for root in ("/usr/share/aws", "/usr/lib/spark", "/usr/lib", "/opt"):
+            if os.path.isdir(root):
+                hits += glob.glob(os.path.join(root, "**", pattern + ".jar"), recursive=True)
+        print(f"PROBE pattern {pattern}: {len(hits)} hit(s)")
+        for h in sorted(set(hits))[:25]:
+            print(f"PROBE     {h}")
+
+    section("classpath resolution")
+    # Ask the JVM whether the Iceberg classes are already reachable without any
+    # spark.jars: if they are, the extension config alone is enough.
+    for cls in (
+        "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
+        "org.apache.iceberg.aws.glue.GlueCatalog",
+        "software.amazon.s3tables.iceberg.S3TablesCatalog",
+    ):
+        try:
+            from pyspark.sql import SparkSession
+            spark = SparkSession.builder.appName("probe").getOrCreate()
+            jvm = spark.sparkContext._jvm
+            loader = jvm.java.lang.Thread.currentThread().getContextClassLoader()
+            try:
+                jvm.java.lang.Class.forName(cls, False, loader)
+                print(f"PROBE class {cls}: PRESENT")
+            except Exception:
+                print(f"PROBE class {cls}: NOT on default classpath")
+        except Exception as e:  # noqa: BLE001
+            print(f"PROBE class {cls}: check failed ({type(e).__name__}: {e})")
+
+    section("iceberg version hint")
+    try:
+        out = subprocess.run(
+            ["bash", "-lc", "ls /usr/share/aws/iceberg/lib 2>/dev/null || true"],
+            capture_output=True, text=True, timeout=30,
+        )
+        print(f"PROBE ls: {out.stdout.strip() or '(empty)'}")
+    except Exception as e:  # noqa: BLE001
+        print(f"PROBE ls failed: {e}")
+
+    print("\nPROBE done")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/aws/run_emr_serverless.py

@@ -54,7 +54,17 @@
 
 # Iceberg ships inside the EMR image; this is a local path in the container, not
 # a download. See the EMR Serverless "Using Apache Iceberg" documentation.
-ICEBERG_JAR = "/usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar"
+#
+# The documented path is the default below. It is overridable because the layout
+# is release-dependent: this exact value does not exist on emr-spark-8.0.0 (the
+# jar name carries "spark3", and EMR 8 runs Spark 4), where the job fails with
+# NoSuchFileException. Set ICEBERG_JAR_PATH to the real path for your release,
+# or to an empty string to omit spark.jars entirely when Iceberg is already on
+# the default classpath. Run with PROBE=1 to have the image report its layout.
+ICEBERG_JAR = os.environ.get(
+    "ICEBERG_JAR_PATH", "/usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar"
+)
+PROBE = os.environ.get("PROBE", "").lower() in ("1", "true", "yes")
 
 # Catalog implementations per storage mode.
 GLUE_CATALOG_IMPL = "org.apache.iceberg.aws.glue.GlueCatalog"
@@ -141,10 +151,12 @@ def _wait_for(get_state, want: set, bad: set, what: str, timeout_s: int = 600) -
 def spark_submit_params(mode: str, catalog_impl: str, warehouse: str) -> str:
     jars = ICEBERG_JAR
     if mode == "s3tables" and S3TABLES_EXTRA_JARS:
-        jars = f"{jars},{S3TABLES_EXTRA_JARS}"
+        jars = ",".join(j for j in (jars, S3TABLES_EXTRA_JARS) if j)
 
-    params = [
-        f"--conf spark.jars={jars}",
+    params = []
+    if jars:
+        params.append(f"--conf spark.jars={jars}")
+    params += [
         "--conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
         "--conf spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog",
         f"--conf spark.sql.catalog.local.catalog-impl={catalog_impl}",
@@ -159,6 +171,43 @@ def spark_submit_params(mode: str, catalog_impl: str, warehouse: str) -> str:
     return " ".join(params)
 
 
+def run_probe(app_id: str, probe_uri: str) -> int:
+    """Submit the diagnostic job and print where its output landed.
+
+    Cheap way to settle release-dependent questions (jar path, whether Iceberg is
+    already on the classpath, whether the S3 Tables catalog exists) instead of
+    guessing across job runs. Output goes to the driver stdout log in S3.
+    """
+    print("\n[driver] === probe: reporting the image layout ===")
+    resp = emr.start_job_run(
+        applicationId=app_id,
+        executionRoleArn=JOB_ROLE_ARN,
+        name=f"{RESOURCE_PREFIX}-probe"[:64],
+        executionTimeoutMinutes=15,
+        jobDriver={"sparkSubmit": {"entryPoint": probe_uri, "entryPointArguments": [],
+                                   "sparkSubmitParameters": ""}},
+        configurationOverrides={
+            "monitoringConfiguration": {
+                "s3MonitoringConfiguration": {"logUri": s3_uri(ENGINE, "logs", RUN_TAG) + "/"}
+            }
+        },
+        tags={"project": "iceberg-matrix", "run": RUN_TAG, "mode": "probe"},
+    )
+    job_id = resp["jobRunId"]
+    state = _wait_for(
+        lambda: emr.get_job_run(applicationId=app_id, jobRunId=job_id)["jobRun"]["state"],
+        want={"SUCCESS", "FAILED", "CANCELLED"}, bad=set(),
+        what=f"probe job {job_id}", timeout_s=1200,
+    )
+    log_prefix = f"{ENGINE}/logs/{RUN_TAG}/applications/{app_id}/jobs/{job_id}/"
+    print(f"[driver] probe {state}")
+    print(f"[driver] read the PROBE lines from the driver stdout under:")
+    print(f"[driver]   s3://{DATA_BUCKET}/{log_prefix}SPARK_DRIVER/stdout.gz")
+    print(f"[driver] e.g. aws s3 cp s3://{DATA_BUCKET}/{log_prefix}SPARK_DRIVER/stdout.gz - "
+          "| gunzip | grep '^PROBE'")
+    return 0 if state == "SUCCESS" else 1
+
+
 def run_mode(app_id: str, mode: str, bundle_uri: str, entry_uri: str) -> dict:
     if mode == "s3buckets":
         catalog_impl = GLUE_CATALOG_IMPL
@@ -294,6 +343,16 @@ def main() -> int:
     modes = ["s3buckets", "s3tables"] if MODES == "both" else [MODES]
     print(f"[driver] region={REGION} bucket={DATA_BUCKET} modes={modes}")
 
+    if PROBE:
+        probe_uri = upload(Path(__file__).with_name("emr_probe.py"),
+                           f"{ENGINE}/scripts/{RUN_TAG}/emr_probe.py")
+        app_id = create_application()
+        Path("/tmp/emr-application-id").write_text(app_id)
+        if os.environ.get("GITHUB_ENV"):
+            with open(os.environ["GITHUB_ENV"], "a") as f:
+                f.write(f"EMR_APPLICATION_ID={app_id}\n")
+        return run_probe(app_id, probe_uri)
+
     bundle = build_bundle(Path("/tmp") / f"{RUN_TAG}-bundle.zip")
     bundle_uri = upload(bundle, f"{ENGINE}/scripts/{RUN_TAG}/bundle.zip")
     entry_uri = upload(Path(__file__).with_name("emr_entrypoint.py"),