[digvijay-y]

What this PR does / why we need it:
Adds a runnable local development example demonstrating SparkClient with Apache Iceberg and MinIO as S3-compatible object storage.

Includes:

• iceberg_minio.py — end-to-end: create namespace, table, write and read data
• docker-compose-iceberg-minio.yml — one-command local setup
• Updated README.md with setup instructions

Tested with:

• iceberg-spark-runtime 1.9.1
• AWS SDK v2 (software.amazon.awssdk:bundle:2.26.24)
• PySpark 3.5.0
• MinIO + tabulario/iceberg-rest

Which issue(s) this PR fixes (optional, in Fixes #<issue number>, #<issue number>, ... format, will close the issue(s) when PR gets merged):

Fixes #500

Checklist:

• Docs included if any changes are user facing

[google-oss-prow]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign electronic-waste for approval. For more information see the Kubernetes Code Review Process.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[Copilot]

Pull request overview

Note

Copilot was unable to run its full agentic suite in this review.

Adds a local-development example for using Spark with Apache Iceberg backed by MinIO (S3-compatible) and an Iceberg REST catalog.

Changes:

• Added a runnable PySpark example that creates, writes, and reads an Iceberg table using MinIO + REST catalog.
• Added a docker-compose stack to spin up MinIO, initialize a warehouse bucket, and run the Iceberg REST catalog.
• Documented setup/run/config/teardown steps in the Spark examples README.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 6 comments.

File	Description
examples/spark/iceberg_minio.py	New end-to-end Iceberg + MinIO example script with Spark session configuration and basic table operations
examples/spark/docker-compose-iceberg-minio.yml	New docker-compose stack for MinIO + bucket init + Iceberg REST catalog
examples/spark/README.md	Documents how to run the Iceberg + MinIO example and its configuration

[tariq-hasan]

Hi @digvijay-y! Thanks for raising the PR. This is a great start. I left a few high-level comments for now. We can go deeper on the review once we align on the approach.

[tariq-hasan]

The test uses pyspark directly. We need to run the test through the Kubeflow Spark SDK.

The value of this example for #470 is proving the Iceberg path works through the SDK's SparkConnect path end-to-end. A local SparkSession validates Iceberg+Spark but exercises no Kubeflow code, so it can't go in the example harness.

[tariq-hasan]

The example needs to be integrated with the test harness to verify that it works as part of the e2e.

[tariq-hasan]

/ok-to-test
/retest

[tariq-hasan]

Hi @digvijay-y! Thanks for implementing the changes.

Please also rebase the PR to fix the merge conflicts.

[tariq-hasan]

The package sizes may be causing a timeout on the E2E tests. Better to replace them with lightweight minimal packages that showcase the end-to-end flow.

[tariq-hasan]

The access key and secret key credentials are best not exposed in spark configs.

[review-notebook-app]

Check out this pull request on 

See visual diffs & provide feedback on Jupyter Notebooks.

---

Powered by ReviewNB

[tariq-hasan]

Thanks @digvijay-y! Let's convert the example into a notebook to match the other refactored Spark examples: https://github.com/kubeflow/sdk/tree/main/examples/spark.

[tariq-hasan]

Hi @digvijay-y! Thanks for making the changes. I have added a few more comments.

[tariq-hasan]

Please remove all changes from hack/e2e-setup-cluster.sh. This file should not be modified.

[tariq-hasan]

Please move this script to hack/setup-iceberg-minio.sh. We no longer maintain a test directory on main.

[tariq-hasan]

No need to configure user and password. Also let's reference the hack directory since we are moving the script.

Suggested change

	.PHONY: test-e2e-setup-fixtures
	test-e2e-setup-fixtures: ## Setup optional Spark E2E fixtures
	@SPARK_TEST_NAMESPACE=$(SPARK_TEST_NAMESPACE) \
	MINIO_ROOT_USER=$(MINIO_ROOT_USER) \
	MINIO_ROOT_PASSWORD=$(MINIO_ROOT_PASSWORD) \
	./test/e2e/spark/fixtures/setup-iceberg-minio.sh
	.PHONY: test-e2e-setup-fixtures
	test-e2e-setup-fixtures: ## Deploy optional Spark E2E fixtures (MinIO + Iceberg REST catalog) into an existing cluster
	@echo "Setting up Spark E2E fixtures..."
	@SPARK_TEST_NAMESPACE=$(SPARK_TEST_NAMESPACE) \
	./hack/setup-iceberg-minio.sh

[tariq-hasan]

Let's remove this.

Suggested change

SPARK_E2E_ENABLE_ICEBERG_MINIO: "1"

[tariq-hasan]

Instead let's add an additional step here.

      - name: Setup Iceberg + MinIO fixtures
        run: |
          make test-e2e-setup-fixtures
        env:
          SPARK_TEST_NAMESPACE: spark-test
        timeout-minutes: 10

[tariq-hasan]

No need to set environment variables for the AWS settings. We still need to configure them as spark config though.

import os
import uuid

from kubeflow.common.types import KubernetesBackendConfig
from kubeflow.spark import Name, SparkClient

namespace = os.environ.get("SPARK_TEST_NAMESPACE", "default")

# Service names resolve from the Spark Connect server pod in the same namespace.
minio_endpoint = os.environ.get("MINIO_ENDPOINT", "http://minio:9000")
iceberg_rest_uri = os.environ.get("ICEBERG_REST_URI", "http://iceberg-rest:8181")
warehouse = os.environ.get("ICEBERG_WAREHOUSE", "s3://warehouse/")

# Local development defaults only. Never commit real credentials to a notebook.
minio_access_key = os.environ.get("MINIO_ACCESS_KEY", "minioadmin")
minio_secret_key = os.environ.get("MINIO_SECRET_KEY", "minioadmin")

client = SparkClient(backend_config=KubernetesBackendConfig(namespace=namespace))
session_name = f"iceberg-minio-{uuid.uuid4().hex[:8]}"

print(f"Namespace: {namespace}")
print(f"MinIO endpoint: {minio_endpoint}")
print(f"Iceberg REST catalog: {iceberg_rest_uri}")

[tariq-hasan]

Let's drop the timeouts and parameterize the fields that go in spark.jars.

iceberg_version = "1.11.0"
maven = "https://repo1.maven.org/maven2/org/apache/iceberg"

spark = client.connect(
    spark_conf={
        "spark.jars": ",".join(
            [
                f"{maven}/iceberg-spark-runtime-4.0_2.13/{iceberg_version}/"
                f"iceberg-spark-runtime-4.0_2.13-{iceberg_version}.jar",
                f"{maven}/iceberg-aws-bundle/{iceberg_version}/"
                f"iceberg-aws-bundle-{iceberg_version}.jar",
            ]
        ),
        "spark.sql.extensions": (
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"
        ),
        "spark.sql.catalog.lakehouse": "org.apache.iceberg.spark.SparkCatalog",
        "spark.sql.catalog.lakehouse.type": "rest",
        "spark.sql.catalog.lakehouse.uri": iceberg_rest_uri,
        "spark.sql.catalog.lakehouse.warehouse": warehouse,
        "spark.sql.catalog.lakehouse.io-impl": "org.apache.iceberg.aws.s3.S3FileIO",
        "spark.sql.catalog.lakehouse.s3.endpoint": minio_endpoint,
        "spark.sql.catalog.lakehouse.s3.path-style-access": "true",
        "spark.sql.catalog.lakehouse.client.region": "us-east-1",
        "spark.sql.catalog.lakehouse.s3.access-key-id": minio_access_key,
        "spark.sql.catalog.lakehouse.s3.secret-access-key": minio_secret_key,
    },
    options=[Name(session_name)],
)
print(f"Connected Spark session: {session_name}")

[tariq-hasan]

Let's cleanup before session deletion.

spark.sql("DROP TABLE IF EXISTS lakehouse.demo.users PURGE")
spark.stop()
client.delete_session(session_name)
print("Iceberg + MinIO example complete.")

[tariq-hasan]

Let's be more thorough in the table metadata inspection and write some assertions since this runs in E2E tests.

snapshots = spark.sql("SELECT snapshot_id, operation FROM lakehouse.demo.users.snapshots").collect()
print(f"Snapshots: {[(s.snapshot_id, s.operation) for s in snapshots]}")
assert len(snapshots) == 1 and snapshots[0].operation == "append", snapshots

data_files = [
    row.file_path for row in spark.sql("SELECT file_path FROM lakehouse.demo.users.files").collect()
]
print("Data files:")
for path in data_files:
    print(f"  {path}")
assert data_files, "Expected at least one data file"
assert all(path.startswith(warehouse) for path in data_files), data_files

[digvijay-y]

@tariq-hasan it looks good now, can you check it whenever possible