feat: first spark attempt

Author: sweb

demos/demos-v2.yaml

@@ -314,20 +314,23 @@ demos:
         description: Git branch to download demo files from
         default: main
   openlineage-trino-airflow:
-    description: Showcase OpenLineage data lineage from Trino and Airflow, visualized in Marquez
+    description: Showcase OpenLineage data lineage from Trino, Airflow and Spark over a shared Iceberg lakehouse, visualized in Marquez
     documentation: https://docs.stackable.tech/home/stable/demos/openlineage-trino-airflow
     stackableStack: openlineage
     labels:
       - trino
       - airflow
+      - spark
       - openlineage
       - marquez
       - data-lineage
+      - iceberg
+      - minio
     manifests:
       - plainYaml: demos/openlineage-trino-airflow/rbac.yaml
       - plainYaml: demos/openlineage-trino-airflow/enable-and-run-dag.yaml
     supportedNamespaces: []
     resourceRequests:
-      cpu: 5000m
-      memory: 12Gi
-      pvc: 15Gi
+      cpu: 10500m
+      memory: 22Gi
+      pvc: 40Gi # 10Gi MinIO + 3x8Gi PostgreSQL; data is tiny (tpch.tiny)

demos/openlineage-trino-airflow/rbac.yaml

@@ -1,6 +1,9 @@
 ---
-# Allow the enable-and-run Job (running under the namespace's default ServiceAccount) to wait for the
-# Airflow StatefulSets to become ready via `kubectl rollout status`.
+# Grants two things to every ServiceAccount in the namespace (see the ClusterRoleBinding below):
+#   * the enable-and-run Job waits for the Airflow StatefulSets via `kubectl rollout status`
+#     (pods + statefulsets get/list/watch), and
+#   * the Airflow DAG's inlined SparkKubernetesOperator/Sensor creates and polls the SparkApplication
+#     (sparkapplications create/get/list/watch).
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:
@@ -22,6 +25,15 @@ rules:
       - get
       - list
       - watch
+  - apiGroups:
+      - spark.stackable.tech
+    resources:
+      - sparkapplications
+    verbs:
+      - get
+      - list
+      - watch
+      - create
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding

docs/modules/demos/pages/openlineage-trino-airflow.adoc

@@ -1,14 +1,17 @@
 = openlineage-trino-airflow
-:description: Demo showcasing OpenLineage data lineage emitted by Trino and Airflow and visualized in Marquez, enabled with minimal configuration.
+:description: Demo showcasing OpenLineage data lineage emitted by Trino, Airflow and Spark over a shared Iceberg lakehouse and visualized in Marquez, enabled with minimal configuration.
 
 :openlineage: https://openlineage.io/
 :marquez: https://marquezproject.ai/
 :trino-openlineage-docs: https://trino.io/docs/current/admin/event-listeners-openlineage.html
 :airflow-openlineage-docs: https://airflow.apache.org/docs/apache-airflow-providers-openlineage/stable/index.html
+:spark-openlineage-docs: https://openlineage.io/docs/integrations/spark/
 
-This demo shows how two Stackable Data Platform (SDP) products - Apache Trino and Apache Airflow - can emit
-{openlineage}[OpenLineage{external-link-icon}^] events with a single configuration toggle each, and how those events
-come together in {marquez}[Marquez{external-link-icon}^], the OpenLineage reference metadata server and UI.
+This demo shows how three Stackable Data Platform (SDP) products - Apache Trino, Apache Airflow and Apache Spark - can
+emit {openlineage}[OpenLineage{external-link-icon}^] events with a single configuration toggle each, and how those events
+come together in {marquez}[Marquez{external-link-icon}^], the OpenLineage reference metadata server and UI. Trino and
+Spark share an Apache Iceberg lakehouse on MinIO (S3), so a table written by Trino and read by Spark links the two
+engines' lineage into one graph.
 
 Install this demo on an existing Kubernetes cluster:
 
@@ -33,9 +36,9 @@ This demo should not be run alongside other demos.
 
 To run this demo, your system needs at least:
 
-* 5 https://kubernetes.io/docs/tasks/debug/debug-cluster/resource-metrics-pipeline/#cpu[cpu units{external-link-icon}^] (core/hyperthread)
-* 12GiB memory
-* 15GiB disk storage
+* 10 https://kubernetes.io/docs/tasks/debug/debug-cluster/resource-metrics-pipeline/#cpu[cpu units{external-link-icon}^] (core/hyperthread)
+* 24GiB memory
+* 128GiB disk storage
 
 == Overview
 
@@ -47,24 +50,44 @@ This demo will
    listener so that every query emits table- and column-level lineage.
 ** *Airflow*: A workflow orchestrator. The native `apache-airflow-providers-openlineage` provider is enabled so that
    DAG and task runs - including the datasets they read and write - are reported as lineage.
-** *PostgreSQL*: Metadata database for Airflow.
+** *Spark*: A distributed data-processing engine. The `openlineage-spark` listener is enabled via `sparkConf` so that
+   the Spark job's reads and writes are reported as lineage.
+** *Hive metastore* + *MinIO* (S3): back the shared Apache Iceberg `lakehouse` catalog that Trino and Spark both use.
+** *PostgreSQL*: Metadata databases for Airflow and the Hive metastore.
 ** *Marquez*: The OpenLineage reference server + web UI (with its own PostgreSQL), used here as the consuming
    application that collects and visualizes the lineage. Marquez is only part of this demo, not an SDP product.
-* Mount and trigger a single Airflow DAG (`openlineage_demo`) that runs a small ELT pipeline in Trino
+* Mount and trigger a single Airflow DAG (`openlineage_demo`) that runs an ELT pipeline in Trino and then a Spark
+  aggregation job
 * Let you explore the resulting lineage graph in the Marquez UI
 
-The DAG builds a layered set of tables using Trino's built-in `tpch` catalog as the source and the in-memory `memory`
-catalog as the sink:
+The DAG reads from Trino's built-in `tpch` catalog and writes Iceberg tables into the shared `lakehouse` catalog. Trino
+produces `lakehouse.demo.customer_orders`; Spark then reads that same Iceberg table and writes
+`lakehouse.demo.orders_by_nation`:
 
 [source]
 ----
 tpch.tiny.customer ─┐
-                    ├─> memory.demo.customer_orders
+                    ├─(Trino)─> lakehouse.demo.customer_orders ─(Spark)─> lakehouse.demo.orders_by_nation
 tpch.tiny.orders  ──┘
 ----
 
-Both products report to the *same* Marquez instance (`http://marquez:5000`): Trino emits the per-query dataset lineage,
-while Airflow emits the job/task run events (and dataset lineage via the Trino SQL operator).
+All three products report to the *same* Marquez instance (`http://marquez:5000`): Trino emits per-query dataset lineage
+(from its event listener), Airflow emits the DAG/task run events _plus_ the dataset lineage of every step, and Spark
+emits the read/write lineage of its job.
+
+The single, connected end-to-end graph lives in the *Airflow* namespace (`airflow-openlineage-demo`). Airflow is the
+orchestrator and a single OpenLineage producer, so it names every dataset consistently: the Trino SQL tasks report
+`tpch.tiny.*` -> `lakehouse.demo.customers/orders` -> `lakehouse.demo.customer_orders`, and the Spark task is made to
+declare its own input/output (`customer_orders` -> `orders_by_nation`) under the *same* dataset identities. That shared
+`lakehouse.demo.customer_orders` node is what joins the Trino ELT and the Spark step into one graph.
+
+[NOTE]
+====
+Marquez keys datasets by `namespace + name`, and each engine names the *same* physical Iceberg table differently
+(Trino's listener by its coordinator URI, Spark by the S3 location / Hive metastore). So Trino's own event-listener
+lineage and Spark's own job lineage show up as separate, self-contained views next to the unified Airflow graph - each
+producer's perspective on the same pipeline. The connected end-to-end flow is the Airflow one.
+====
 
 == How OpenLineage is enabled
 
@@ -95,6 +118,81 @@ AIRFLOW__OPENLINEAGE__NAMESPACE="airflow-openlineage-demo"
 
 See the {airflow-openlineage-docs}[Airflow OpenLineage provider docs{external-link-icon}^] for details.
 
+*Spark* loads the `openlineage-spark` listener and points it at Marquez purely through `sparkConf` on the
+`SparkApplication` (plus the package on the classpath via `deps.packages`):
+
+[source,properties]
+----
+spark.extraListeners=io.openlineage.spark.agent.OpenLineageSparkListener
+spark.openlineage.transport.type=http
+spark.openlineage.transport.url=http://marquez:5000
+spark.openlineage.namespace=spark-openlineage-demo
+spark.app.name=orders-by-nation
+# parent* are injected at task runtime by the Airflow operator (see below) so they carry the
+# current run's identity - they link the Spark run to its Airflow task:
+spark.openlineage.parentJobNamespace=airflow-openlineage-demo
+spark.openlineage.parentJobName=openlineage_demo.spark_orders_by_nation
+spark.openlineage.parentRunId=<this Airflow task's OpenLineage run id>
+----
+
+See the {spark-openlineage-docs}[OpenLineage Spark integration docs{external-link-icon}^] for details.
+
+[NOTE]
+====
+The Stackable `SparkKubernetesOperator`/`Sensor` are not bundled in the Airflow image, so the demo DAG inlines a small
+in-cluster implementation. It does three things worth calling out, each a lever for cleaner lineage:
+
+* It creates the `SparkApplication` *and waits for it in the same task* (no separate sensor task, so no extra empty
+  "monitor" job in the graph).
+* It sets `spark.openlineage.parentJobNamespace/parentJobName/parentRunId` from the running task's OpenLineage identity
+  (via the provider's `lineage_run_id`/`lineage_job_name` macros). This makes the Spark run report the Airflow task as
+  its *root*, so the real Spark job is reachable from the DAG in Marquez.
+* It implements `get_openlineage_facets_on_complete` to declare the Spark step's input (`customer_orders`) and output
+  (`orders_by_nation`) under the same dataset identities Airflow uses for the Trino tables. Airflow cannot introspect
+  the out-of-process Spark job, so this explicit declaration is what draws the edge into the ELT graph.
+====
+
+=== Getting clean, connected lineage
+
+Enabling the listeners is only half the story - a few extra choices make the graph readable and connected. The demo
+applies them so it can double as a reference:
+
+[cols="1,2"]
+|===
+| Lever | Effect
+
+| `spark.openlineage.parent*` (from the Airflow task's run id)
+| Links the Spark run to its Airflow task (the Spark run's `root` becomes the task), so the orchestrated Spark job is
+  reachable from the DAG rather than floating in its own namespace.
+
+| `get_openlineage_facets_on_complete` on the Spark operator
+| Declares the Spark step's datasets under the Trino tables' identities, joining Spark to the ELT graph in the Airflow
+  namespace.
+
+| `CREATE OR REPLACE TABLE` (one statement per task, not `DROP` + `CREATE`)
+| One SQL statement per task means one lineage-bearing job, instead of an extra empty job for the throwaway `DROP`.
+
+| One operator that submits *and* waits
+| Avoids a second, dataset-less sensor task appearing as an empty node.
+
+| `spark.app.name` as a Spark *config* (not only `SparkSession.builder.appName`)
+| The listener reads the app name at `SparkContext` start; setting it in config gives the Spark jobs a stable name.
+|===
+
+[NOTE]
+====
+Some rough edges remain and are *inherent to today's OpenLineage integrations*, not misconfiguration:
+
+* Airflow's SQL provider emits an empty *task* job plus a `.query.N` sub-job that carries the datasets - the datasets
+  live on the sub-job.
+* `openlineage-spark` (in k8s cluster mode) still names its internal application-root job `unknown` and leaves
+  `spark_applicationDetails.applicationName` null, even with `spark.app.name` set.
+* The DAG-level `openlineage_demo` job and the `create_schema` (DDL) job have no datasets by nature.
+
+These do not affect the correctness of the data flow; they are visible artifacts of how the current integrations model
+jobs.
+====
+
 == List deployed Stackable services
 
 To list the installed Stackable services run the following command:
@@ -112,11 +210,15 @@ $ stackablectl stacklet list --namespace openlineage_demo
 └─────────┴─────────┴───────────────────┴──────────────────────────────────────────┴─────────────────────────────────┘
 ----
 
+The `hive` metastore also appears as a stacklet.
+
 include::partial$instance-hint.adoc[]
 
 [NOTE]
 ====
-Marquez is not an SDP product, so it does not appear in `stackablectl stacklet list`.
+Marquez, MinIO and the Spark job are not shown by `stackablectl stacklet list`: Marquez and MinIO are not SDP products,
+and the Spark job runs as a short-lived `SparkApplication` submitted by the Airflow DAG rather than a long-running
+stacklet.
 ====
 
 == Explore the lineage in Marquez
@@ -132,29 +234,55 @@ Then open http://localhost:5000 in your browser.
 
 Use the namespace selector at the top to switch between the namespaces that produced the lineage:
 
-* the Trino namespace (named after `openlineage-event-listener.trino.uri`, i.e. `http://trino-coordinator:8080`),
-  which contains the `tpch.tiny.*` and `memory.demo.*` datasets and the queries connecting them, and
-* `airflow-openlineage-demo`, which contains the `openlineage_demo` DAG and its tasks.
+* `airflow-openlineage-demo` - the *connected end-to-end graph*. It contains the `openlineage_demo` DAG, its tasks, and
+  the datasets `tpch.tiny.*`, `lakehouse.demo.customers/orders/customer_orders` and `lakehouse.demo.orders_by_nation`.
+* the Trino namespace (named after `openlineage-event-listener.trino.uri`, i.e. `http://trino-coordinator:8080`) - the
+  Trino event listener's own view, with column-level lineage for each query.
+* `spark-openlineage-demo` (plus `s3://lakehouse` and `hive://...`) - the Spark job's own view of its read/write.
+
+To see the full flow, open the `lakehouse.demo.orders_by_nation` dataset (or the `openlineage_demo.spark_orders_by_nation`
+task) in the `airflow-openlineage-demo` namespace and expand upstream. You get one connected graph:
 
-Selecting the `memory.demo.customer_orders` dataset and expanding upstream shows the full graph back to
-`tpch.tiny.customer` and `tpch.tiny.orders`, including column-level lineage for the individual queries.
+[source]
+----
+tpch.tiny.customer ─▶ load_customers ─▶ customers ─┐
+                                                   ├─▶ customer_orders ─▶ lakehouse.demo.customer_orders
+tpch.tiny.orders   ─▶ load_orders    ─▶ orders   ──┘                              │
+                                                                                  ▼
+                                            spark_orders_by_nation ─▶ lakehouse.demo.orders_by_nation
+----
+
+The real Spark job is not lost: because of the parent-run link, the Spark run reports the
+`openlineage_demo.spark_orders_by_nation` task as its *root*, so you can drill from that task into the Spark run itself
+(in the `spark-openlineage-demo` namespace).
+
+[TIP]
+====
+Open a *dataset* or a *task* node - for example `spark_orders_by_nation` or the `orders_by_nation` dataset - not the
+top-level `openlineage_demo` DAG node. Airflow reports the DAG-level job with no inputs/outputs of its own, so that one
+node looks empty; the data flow hangs off the task and dataset nodes.
+====
 
 [NOTE]
 ====
-Trino and Airflow name datasets independently, so a given table may appear under both the Trino namespace (from the
-Trino event listener) and the Airflow namespace (from the SQL operator's extractor). This is expected - each producer
-reports lineage from its own perspective.
+The Trino event-listener view (`trino://...`) and the Spark job's own view (`spark-openlineage-demo` / `s3://lakehouse`
+/ `hive://...`) are separate, self-contained graphs alongside the Airflow one. Marquez identifies datasets by
+`namespace + name`, and each engine names the same physical Iceberg table differently (Trino by its coordinator URI,
+Spark by the storage location / metastore), so they do not auto-merge. That is expected: each producer reports lineage
+from its own perspective, and the *connected* end-to-end view is the Airflow one, where a single producer (the
+orchestrator) names every dataset consistently.
 ====
 
 == Summary
 
-This demo showed that both Trino and Airflow can emit OpenLineage data to Marquez with only a small amount of
-configuration and no changes to the SDP operators or CRDs. It serves as a validation of the approach ahead of adding
-first-class OpenLineage support to the platform.
+This demo showed that Trino, Airflow and Spark can all emit OpenLineage data to Marquez with only a small amount of
+configuration and no changes to the SDP operators or CRDs, and that the Airflow orchestrator can tie a Trino ELT and a
+Spark job - sharing an Iceberg lakehouse - into a single connected lineage graph. It serves as a validation of the
+approach ahead of adding first-class OpenLineage support to the platform.
 
 == Where to go from here
 
 * Trigger the `openlineage_demo` DAG again from the Airflow UI and watch new runs appear in Marquez.
-* Run your own queries against Trino (for example via `stackablectl stacklet list` to find the coordinator endpoint)
-  and see them show up as additional lineage.
+* Query the Spark output from Trino: `SELECT * FROM lakehouse.demo.orders_by_nation ORDER BY total_spent DESC` (use
+  `stackablectl stacklet list` to find the coordinator endpoint) and see the query show up as additional lineage.
 * Read more about {openlineage}[OpenLineage{external-link-icon}^] and {marquez}[Marquez{external-link-icon}^].