diff --git a/demos/demos-v2.yaml b/demos/demos-v2.yaml index e68c301d..a4b1dc9f 100644 --- a/demos/demos-v2.yaml +++ b/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) diff --git a/demos/openlineage-trino-airflow/rbac.yaml b/demos/openlineage-trino-airflow/rbac.yaml index f29276c9..5fadaeed 100644 --- a/demos/openlineage-trino-airflow/rbac.yaml +++ b/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 diff --git a/docs/modules/demos/pages/openlineage-trino-airflow.adoc b/docs/modules/demos/pages/openlineage-trino-airflow.adoc index c1be3c69..a8eb32c6 100644 --- a/docs/modules/demos/pages/openlineage-trino-airflow.adoc +++ b/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 @@ -100,6 +123,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= +---- + +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: @@ -117,11 +215,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. ==== == Verify the demo worked @@ -205,13 +307,14 @@ Airflow contributes the task/DAG structure around the same tables. == 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}^]. diff --git a/stacks/openlineage/airflow.yaml b/stacks/openlineage/airflow.yaml index ede5c2ed..a9789b1a 100644 --- a/stacks/openlineage/airflow.yaml +++ b/stacks/openlineage/airflow.yaml @@ -57,6 +57,9 @@ spec: # Trino connection used by the demo DAG. TLS verification is disabled because Trino uses an # auto-generated (self-signed) CA in this demo; the password matches the `trino-users` Secret. AIRFLOW_CONN_TRINO_DEFAULT: '{"conn_type": "trino", "host": "trino-coordinator", "port": 8443, "login": "admin", "password": "{{ trinoAdminPassword }}", "extra": {"protocol": "https", "verify": false}}' + # In-cluster Kubernetes connection used by the inlined SparkKubernetesOperator/Sensor in the DAG + # to create and poll the SparkApplication. Same encoded connection as the airflow stack uses. + AIRFLOW_CONN_KUBERNETES_IN_CLUSTER: "kubernetes://?__extra__=%7B%22extra__kubernetes__in_cluster%22%3A+true%2C+%22extra__kubernetes__kube_config%22%3A+%22%22%2C+%22extra__kubernetes__kube_config_path%22%3A+%22%22%2C+%22extra__kubernetes__namespace%22%3A+%22%22%7D" configOverrides: webserver_config.py: # Allow "POST /login/" without CSRF token @@ -91,77 +94,327 @@ metadata: data: openlineage_demo.py: | """ - A minimal ELT pipeline against Trino that produces clean, layered lineage: + A layered ELT + Spark pipeline that produces cross-engine lineage in Marquez: tpch.tiny.customer ─┐ - ├─> memory.demo.customer_orders + ├─(Trino)─> lakehouse.demo.customer_orders ─(Spark)─> lakehouse.demo.orders_by_nation tpch.tiny.orders ──┘ - Airflow emits task/run + dataset lineage (via the Trino SQL operator) and Trino emits - per-query table-/column-level lineage. Both land in the same Marquez instance. + The Trino steps read from the built-in `tpch` catalog and write Iceberg tables into the + `lakehouse` catalog (shared hive-iceberg metastore + MinIO). The final step submits a Spark job + that reads `lakehouse.demo.customer_orders` and writes `lakehouse.demo.orders_by_nation` into the + *same* Iceberg catalog. + + The single, connected end-to-end graph is the *Airflow* one: Airflow is one producer and names + every dataset consistently, and the Spark step is stitched in by declaring its I/O under the same + identities (see get_openlineage_facets_on_complete below). Trino's own event-listener lineage and + Spark's own listener lineage are separate, self-contained views (each engine names datasets its + own way, so Marquez does not auto-merge them). All producers report to one Marquez. + + The CLEAN-LINEAGE KNOB comments below call out what makes the graph readable and connected. + + NOTE: the Stackable `SparkKubernetesOperator`/`Sensor` from the `stackable.*` package are not in + the Airflow image, so a minimal in-cluster implementation is inlined below. It creates the + SparkApplication CR *and* waits for it (one operator, so there is no extra empty "sensor" job). """ - from datetime import datetime + import time + from datetime import datetime, timedelta + from typing import TYPE_CHECKING, Optional, Sequence from airflow import DAG + from airflow.exceptions import AirflowException + from airflow.models import BaseOperator + from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator + if TYPE_CHECKING: + from airflow.utils.context import Context + + + class SparkKubernetesOperator(BaseOperator): + template_fields: Sequence[str] = ("application_file", "namespace") + ui_color = "#f4a460" + + FAILURE_STATES = ("Failed",) + SUCCESS_STATES = ("Succeeded",) + + def __init__( + self, + *, + application_file: dict, + namespace: Optional[str] = None, + kubernetes_conn_id: str = "kubernetes_in_cluster", + api_group: str = "spark.stackable.tech", + api_version: str = "v1alpha1", + openlineage_inputs: Optional[list] = None, + openlineage_outputs: Optional[list] = None, + poll_interval: int = 10, + timeout: int = 1800, + **kwargs, + ) -> None: + super().__init__(**kwargs) + self.application_file = application_file + self.namespace = namespace + self.kubernetes_conn_id = kubernetes_conn_id + self.api_group = api_group + self.api_version = api_version + self.plural = "sparkapplications" + self.poll_interval = poll_interval + self.timeout = timeout + # (namespace, name) pairs the *Airflow* OpenLineage provider will report as this task's + # datasets. Set them to the SAME identities Airflow emits for the Trino tables so the Spark + # step joins the ELT flow into one connected graph in the Airflow namespace (Marquez keys + # datasets by namespace+name; matching them is what draws the edge). See the DAG below. + self.openlineage_inputs = openlineage_inputs or [] + self.openlineage_outputs = openlineage_outputs or [] + + # CLEAN-LINEAGE KNOB #1: do NOT hand the Spark job this task's OpenLineage run identity via + # `spark.openlineage.parent*`. It is tempting (it would nest the Spark run under this Airflow + # task), but it actively BREAKS the Spark lineage in Marquez 0.51.x (explained below). + # + # Why: Spark correctly emits ONE application job named `orders_by_nation` (bare) carrying a + # parent facet, and its SQL children reference that same bare parent. But Marquez renames any + # job that has a parent facet to `{parentJobName}.{jobName}` UNLESS the name already starts + # with the parent. The SQL jobs (`orders_by_nation..`) already do, so they're left + # alone; the application job (`orders_by_nation`) does not start with its parent + # (`openlineage_demo.spark_orders_by_nation`), so Marquez stores it as + # `openlineage_demo.spark_orders_by_nation.orders_by_nation` while the children still point at + # bare `orders_by_nation` -> the one run splits into two app nodes and the chain breaks. + # + # Marquez draws only job<->dataset edges (never job->job parent edges), so the parent facet buys + # no visible edge anyway. Airflow<->Spark is stitched via shared DATASET identities instead, in + # get_openlineage_facets_on_complete (KNOB #2). `spark.openlineage.appName` (KNOB #3, in the + # SparkApplication) still pins a stable, non-`unknown` job name. + + def execute(self, context: "Context"): + hook = KubernetesHook(conn_id=self.kubernetes_conn_id) + name = self.application_file["metadata"]["name"] + self.log.info("Creating SparkApplication %s ...", name) + hook.create_custom_object( + group=self.api_group, + version=self.api_version, + plural=self.plural, + body=self.application_file, + namespace=self.namespace, + ) + # Wait inline (no separate sensor task => no extra empty job in the lineage graph). + deadline = time.monotonic() + self.timeout + while True: + response = hook.get_custom_object( + group=self.api_group, + version=self.api_version, + plural=self.plural, + name=name, + namespace=self.namespace, + ) + phase = response.get("status", {}).get("phase") + if phase in self.FAILURE_STATES: + raise AirflowException(f"SparkApplication {name} failed with state: {phase}") + if phase in self.SUCCESS_STATES: + self.log.info("SparkApplication %s succeeded", name) + return name + if time.monotonic() > deadline: + raise AirflowException(f"SparkApplication {name} timed out (state: {phase})") + self.log.info("SparkApplication %s is in state: %s", name, phase) + time.sleep(self.poll_interval) + + def get_openlineage_facets_on_complete(self, task_instance): + # CLEAN-LINEAGE KNOB #2: emit dataset lineage for the Spark step from Airflow's side. Because + # the SparkApplication runs out-of-process, Airflow can't introspect it - so we declare the + # I/O explicitly. Using the Trino datasets' exact namespace+names makes `customer_orders` the + # shared node between the Trino ELT and this Spark step, connecting the whole flow in one + # Marquez graph. + from airflow.providers.openlineage.extractors import OperatorLineage + from airflow.providers.common.compat.openlineage.facet import Dataset + + return OperatorLineage( + inputs=[Dataset(namespace=ns, name=name) for ns, name in self.openlineage_inputs], + outputs=[Dataset(namespace=ns, name=name) for ns, name in self.openlineage_outputs], + ) + + + def build_spark_application(name: str, namespace: str) -> dict: + """The SparkApplication that reads customer_orders and writes orders_by_nation.""" + return { + "apiVersion": "spark.stackable.tech/v1alpha1", + "kind": "SparkApplication", + "metadata": {"name": name, "namespace": namespace}, + "spec": { + "sparkImage": { + # Iceberg 1.11.0 only supports Spark 4.1.x (kept in sync with the data-lakehouse + # demo). NOTE: OpenLineage's newest Spark module is `spark40` - the listener on + # Spark 4.1 is not officially covered yet; verify events reach Marquez on a real + # cluster and, if the listener breaks, drop to Spark 4.0.x + Iceberg 1.10.x. + "productVersion": "4.1.2", + "pullPolicy": "IfNotPresent", + }, + "mode": "cluster", + "mainApplicationFile": "local:///stackable/spark/jobs/orders-by-nation.py", + "deps": { + "packages": [ + "org.apache.iceberg:iceberg-spark-runtime-4.0_2.13:1.11.0", + "io.openlineage:openlineage-spark_2.13:1.51.0", + ] + }, + "s3connection": {"reference": "minio"}, + "sparkConf": { + # CLEAN-LINEAGE KNOB #3: pin the OpenLineage job name. + # `spark.app.name` alone is NOT enough: the OL listener resolves the job name via + # SparkContext.appName() at context-build time, which in cluster mode (esp. on the + # not-yet-officially-covered Spark 4.1) is intermittently still blank -> it falls back + # to the literal "unknown" (observed: same config produced both `orders_by_nation` and + # `unknown` across runs). `spark.openlineage.appName` is parsed straight from SparkConf + # at listener construction (highest-priority job-name provider, timing-independent), so + # it wins unconditionally. Keep spark.app.name too for the Spark UI / general naming. + "spark.app.name": "orders-by-nation", + "spark.openlineage.appName": "orders-by-nation", + # --- Iceberg catalog `lakehouse` on the shared hive-iceberg metastore --- + "spark.sql.extensions": "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions", + "spark.sql.catalog.lakehouse": "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.lakehouse.type": "hive", + "spark.sql.catalog.lakehouse.uri": "thrift://hive-iceberg-metastore:9083", + # --- OpenLineage: the Spark "toggle" - listener + HTTP transport to Marquez --- + "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", + # NB: spark.openlineage.parent* is deliberately NOT set here or at runtime — see + # CLEAN-LINEAGE KNOB #1 above (it triggers a Marquez job-rename that splits the + # Spark app node). + }, + "volumes": [ + {"name": "script", "configMap": {"name": "spark-openlineage-job"}}, + { + "name": "spark-tls", + "ephemeral": { + "volumeClaimTemplate": { + "metadata": { + "annotations": { + "secrets.stackable.tech/class": "tls", + "secrets.stackable.tech/scope": "pod", + "secrets.stackable.tech/format": "tls-pkcs12", + "secrets.stackable.tech/format.compatibility.tls-pkcs12.password": "changeit", + "secrets.stackable.tech/backend.autotls.cert.lifetime": "14d", + } + }, + "spec": { + "accessModes": ["ReadWriteOnce"], + "resources": {"requests": {"storage": "1"}}, + "storageClassName": "secrets.stackable.tech", + }, + } + }, + }, + ], + "job": {"config": {"resources": {"cpu": {"min": "100m", "max": "1"}}}}, + "driver": { + "config": { + "resources": { + "cpu": {"min": "1", "max": "1"}, + "memory": {"limit": "2Gi"}, + }, + "volumeMounts": [ + {"name": "script", "mountPath": "/stackable/spark/jobs"}, + {"name": "spark-tls", "mountPath": "/stackable/tls"}, + ], + } + }, + "executor": { + "replicas": 1, + "config": { + "resources": { + "cpu": {"min": "1", "max": "2"}, + "memory": {"limit": "2Gi"}, + }, + "volumeMounts": [ + {"name": "script", "mountPath": "/stackable/spark/jobs"}, + {"name": "spark-tls", "mountPath": "/stackable/tls"}, + ], + }, + }, + }, + } + + with DAG( dag_id="openlineage_demo", schedule=None, start_date=datetime(2024, 1, 1), catchup=False, - tags=["openlineage", "trino"], + dagrun_timeout=timedelta(minutes=60), + tags=["openlineage", "trino", "spark", "iceberg"], ) as dag: + # CLEAN-LINEAGE KNOB #4: keep each SQL task a *single* statement. `CREATE OR REPLACE TABLE` + # (instead of DROP + CREATE) means one query per task, so each task maps to exactly one lineage + # job that carries the datasets - no empty extra `.query.N` job for the throwaway DROP. + # + # CREATE SCHEMA is pure DDL with no dataset input/output, so its job is intentionally + # dataset-less (OpenLineage only tracks data flow, not catalog/namespace management). create_schema = SQLExecuteQueryOperator( task_id="create_schema", conn_id="trino_default", - sql="CREATE SCHEMA IF NOT EXISTS memory.demo", + # s3a:// (not s3://) so the location scheme matches what the Spark S3A client uses when it + # reads the table back - avoids a Trino-writes-s3:// / Spark-reads-s3a:// mismatch. + # `demo` is the bucket the standalone MinIO template creates. + sql="CREATE SCHEMA IF NOT EXISTS lakehouse.demo WITH (location = 's3a://demo/lakehouse')", ) load_customers = SQLExecuteQueryOperator( task_id="load_customers", conn_id="trino_default", - sql=[ - "DROP TABLE IF EXISTS memory.demo.customers", - ( - "CREATE TABLE memory.demo.customers AS " - "SELECT custkey, name, nationkey, acctbal " - "FROM tpch.tiny.customer" - ), - ], + sql=( + "CREATE OR REPLACE TABLE lakehouse.demo.customers AS " + "SELECT custkey, name, nationkey, acctbal " + "FROM tpch.tiny.customer" + ), ) load_orders = SQLExecuteQueryOperator( task_id="load_orders", conn_id="trino_default", - sql=[ - "DROP TABLE IF EXISTS memory.demo.orders", - ( - "CREATE TABLE memory.demo.orders AS " - "SELECT orderkey, custkey, orderstatus, totalprice, orderdate " - "FROM tpch.tiny.orders" - ), - ], + sql=( + "CREATE OR REPLACE TABLE lakehouse.demo.orders AS " + "SELECT orderkey, custkey, orderstatus, totalprice, orderdate " + "FROM tpch.tiny.orders" + ), ) customer_orders = SQLExecuteQueryOperator( task_id="customer_orders", conn_id="trino_default", - sql=[ - "DROP TABLE IF EXISTS memory.demo.customer_orders", - ( - "CREATE TABLE memory.demo.customer_orders AS " - "SELECT c.custkey, c.name, c.nationkey, " - " count(o.orderkey) AS order_count, " - " sum(o.totalprice) AS total_spent " - "FROM memory.demo.customers c " - "LEFT JOIN memory.demo.orders o ON c.custkey = o.custkey " - "GROUP BY c.custkey, c.name, c.nationkey" - ), - ], + sql=( + "CREATE OR REPLACE TABLE lakehouse.demo.customer_orders AS " + "SELECT c.custkey, c.name, c.nationkey, " + " count(o.orderkey) AS order_count, " + " sum(o.totalprice) AS total_spent " + "FROM lakehouse.demo.customers c " + "LEFT JOIN lakehouse.demo.orders o ON c.custkey = o.custkey " + "GROUP BY c.custkey, c.name, c.nationkey" + ), + ) + + with open("/run/secrets/kubernetes.io/serviceaccount/namespace") as f: + namespace = f.read().strip() + spark_app_name = "orders-by-nation-" + datetime.utcnow().strftime("%Y%m%d%H%M%S") + + # OpenLineage dataset namespace Airflow uses for Trino tables. It is derived from the + # AIRFLOW_CONN_TRINO_DEFAULT connection (host `trino-coordinator`, port 8443), so it must stay + # in sync with that connection. Declaring the Spark step's datasets under this same namespace + # (and the same table names the Trino SQL tasks produce) is what stitches the Spark step onto + # the ELT graph: `lakehouse.demo.customer_orders` becomes the shared input/output node. + trino_ol_namespace = "trino://trino-coordinator:8443" + + # Single operator: submits the SparkApplication, waits for it, declares its datasets, and links + # the Spark run to this task. No separate sensor => no empty monitor job. + spark_orders_by_nation = SparkKubernetesOperator( + task_id="spark_orders_by_nation", + namespace=namespace, + application_file=build_spark_application(spark_app_name, namespace), + openlineage_inputs=[(trino_ol_namespace, "lakehouse.demo.customer_orders")], + openlineage_outputs=[(trino_ol_namespace, "lakehouse.demo.orders_by_nation")], ) - create_schema >> [load_customers, load_orders] >> customer_orders + create_schema >> [load_customers, load_orders] >> customer_orders >> spark_orders_by_nation --- apiVersion: v1 kind: Secret diff --git a/stacks/openlineage/hive-metastore.yaml b/stacks/openlineage/hive-metastore.yaml new file mode 100644 index 00000000..bf8bc8d1 --- /dev/null +++ b/stacks/openlineage/hive-metastore.yaml @@ -0,0 +1,38 @@ +# Hive metastore backing the Trino *and* Spark Iceberg catalog (both named `lakehouse`). +# +# Sharing a single metastore is what makes the cross-engine lineage work: the table Trino's ELT +# writes (`lakehouse.demo.customer_orders`) is the exact same physical Iceberg table Spark reads, +# so Marquez sees one dataset produced by Trino and consumed by Spark. +# +# The metastore's own metadata DB is the postgresql-hive-iceberg chart deployed by this stack. +--- +apiVersion: hive.stackable.tech/v1alpha1 +kind: HiveCluster +metadata: + name: hive-iceberg +spec: + image: + # Cannot upgrade to 4.0.1 because we run into github.com/apache/iceberg/issues/12878 + # (kept in sync with the data-lakehouse-iceberg-trino-spark stack). + productVersion: 4.0.0 + clusterConfig: + metadataDatabase: + postgresql: + host: postgresql-hive-iceberg + database: hive + credentialsSecretName: hive-postgres-credentials + s3: + reference: minio + metastore: + roleGroups: + default: + replicas: 1 +--- +apiVersion: v1 +kind: Secret +metadata: + name: hive-postgres-credentials +type: Opaque +stringData: + username: hive + password: hive diff --git a/stacks/openlineage/s3-connection.yaml b/stacks/openlineage/s3-connection.yaml new file mode 100644 index 00000000..09743943 --- /dev/null +++ b/stacks/openlineage/s3-connection.yaml @@ -0,0 +1,42 @@ +# S3 (MinIO) connection shared by Trino (Iceberg catalog), the Hive metastore and Spark. +# +# The Spark part of this demo writes an Iceberg table to the `demo` bucket on MinIO and Trino +# reads it back, so both engines resolve table locations through this single S3Connection. MinIO is +# deployed with TLS (see the minio-distributed-small-tls template referenced from the stack), hence +# the `tls` verification block pointing at the `tls` SecretClass that secret-operator ships. +--- +apiVersion: s3.stackable.tech/v1alpha1 +kind: S3Connection +metadata: + name: minio +spec: + host: minio.{{ NAMESPACE }}.svc.cluster.local + port: 9000 + accessStyle: Path + credentials: + secretClass: minio-s3-credentials + tls: + verification: + server: + caCert: + secretClass: tls +--- +apiVersion: secrets.stackable.tech/v1alpha1 +kind: SecretClass +metadata: + name: minio-s3-credentials +spec: + backend: + k8sSearch: + searchNamespace: + pod: {} +--- +apiVersion: v1 +kind: Secret +metadata: + name: minio-s3-credentials + labels: + secrets.stackable.tech/class: minio-s3-credentials +stringData: + accessKey: admin + secretKey: "{{ minioAdminPassword }}" diff --git a/stacks/openlineage/spark.yaml b/stacks/openlineage/spark.yaml new file mode 100644 index 00000000..401c1a6a --- /dev/null +++ b/stacks/openlineage/spark.yaml @@ -0,0 +1,38 @@ +# PySpark job that the Airflow DAG submits as a SparkApplication (see the inlined +# SparkKubernetesOperator in airflow.yaml). +# +# It reads the Iceberg table the Trino ELT produced (`lakehouse.demo.customer_orders`) and writes a +# small aggregate (`lakehouse.demo.orders_by_nation`) back into the same Iceberg catalog. Because the +# OpenLineage Spark listener is enabled via sparkConf on the SparkApplication, this read+write is +# reported to Marquez - the Spark half of the lineage graph, joined to the Trino half through the +# shared `lakehouse.demo.customer_orders` table. +# +# The Iceberg catalog (`lakehouse`), S3 access and the OpenLineage transport are all configured in +# the SparkApplication's sparkConf, not here. +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: spark-openlineage-job +data: + orders-by-nation.py: | + from pyspark.sql import SparkSession + from pyspark.sql.functions import count, sum as sum_ + + spark = SparkSession.builder.appName("orders-by-nation").getOrCreate() + + # Source table written by the Trino ELT steps of the same Airflow DAG. + customer_orders = spark.table("lakehouse.demo.customer_orders") + + orders_by_nation = ( + customer_orders.groupBy("nationkey").agg( + count("custkey").alias("num_customers"), + sum_("order_count").alias("total_orders"), + sum_("total_spent").alias("total_spent"), + ) + ) + + # createOrReplace keeps the demo re-runnable: every DAG run rebuilds the table. + orders_by_nation.writeTo("lakehouse.demo.orders_by_nation").using("iceberg").createOrReplace() + + spark.stop() diff --git a/stacks/openlineage/trino.yaml b/stacks/openlineage/trino.yaml index ec779b91..33140e2c 100644 --- a/stacks/openlineage/trino.yaml +++ b/stacks/openlineage/trino.yaml @@ -6,6 +6,10 @@ # * `podOverrides` mounts the `event-listener.properties` file (from the ConfigMap below). # The `openlineage` event listener plugin itself ships in the Trino distribution since Trino 450. # +# Trino here has two catalogs: `tpch` as the read-only source, and an Iceberg `lakehouse` catalog +# (on the shared hive-iceberg metastore + MinIO) that the demo DAG writes its ELT output into and +# that Spark later reads/extends - giving cross-engine lineage over shared Iceberg tables. +# # For now, on K8s 1.35, TrinoCatalogs need to be deployed before the TrinoCluster. # See: https://github.com/stackabletech/trino-operator/issues/854 --- @@ -19,16 +23,22 @@ spec: connector: tpch: {} --- +# Iceberg catalog backed by the shared `hive-iceberg` metastore and the MinIO `demo` bucket. +# The demo DAG writes its ELT output here as Iceberg tables; the Spark job then reads one of them and +# writes a derived table back - both engines pointing at this same catalog + metastore + bucket. apiVersion: trino.stackable.tech/v1alpha1 kind: TrinoCatalog metadata: - name: memory + name: lakehouse labels: trino: trino spec: connector: - generic: - connectorName: memory + iceberg: + metastore: + configMap: hive-iceberg + s3: + reference: minio --- apiVersion: v1 kind: ConfigMap diff --git a/stacks/stacks-v2.yaml b/stacks/stacks-v2.yaml index 09ee5859..626cd04f 100644 --- a/stacks/stacks-v2.yaml +++ b/stacks/stacks-v2.yaml @@ -685,32 +685,48 @@ stacks: default: main openlineage: description: >- - Stack for the OpenLineage demo: Trino and Airflow both emitting OpenLineage events to Marquez + Stack for the OpenLineage demo: Trino, Airflow and Spark emitting OpenLineage events to Marquez, + with a shared Iceberg lakehouse on MinIO tying the Trino and Spark lineage together stackableRelease: dev stackableOperators: - commons - listener - secret + - hive - trino + - spark-k8s - airflow labels: - trino - airflow + - spark - openlineage - marquez - data-lineage + - iceberg + - minio manifests: - helmChart: stacks/_templates/postgresql-airflow.yaml + # Metadata DB for the hive-iceberg metastore. + - helmChart: stacks/_templates/postgresql-hive-iceberg.yaml + # MinIO (S3, TLS, standalone) holding the `demo` bucket that Trino and Spark share. This demo + # only writes kilobyte-scale tpch.tiny data, so the single-node 10Gi template is plenty - no need + # for the distributed multi-50Gi templates the data-heavy demos use. + - plainYaml: stacks/_templates/minio-tls/rendered-chart.yaml + - plainYaml: stacks/openlineage/s3-connection.yaml + - plainYaml: stacks/openlineage/hive-metastore.yaml # Marquez publishes no Helm repo/OCI chart, so its chart is vendored as rendered plain YAML. # See the header of stacks/openlineage/marquez.yaml for provenance and how to regenerate. - plainYaml: stacks/openlineage/marquez.yaml - plainYaml: stacks/openlineage/trino.yaml + # PySpark job script the Airflow DAG submits as a SparkApplication. + - plainYaml: stacks/openlineage/spark.yaml - plainYaml: stacks/openlineage/airflow.yaml supportedNamespaces: [] resourceRequests: - cpu: 5000m - memory: 12Gi - pvc: 15Gi + cpu: 10500m # ~8.5 cores steady (Trino, Airflow roles, Hive, MinIO, Marquez) + ~2 for the Spark job + memory: 22Gi # measured ~18Gi steady + ~4Gi while the Spark driver/executor run + pvc: 40Gi # 10Gi MinIO + 3x8Gi PostgreSQL (airflow, hive-iceberg, marquez) parameters: - name: trinoAdminPassword description: Password of the Trino admin user @@ -718,3 +734,6 @@ stacks: - name: airflowAdminPassword description: Password of the Airflow admin user default: adminadmin + - name: minioAdminPassword + description: Password of the MinIO admin user + default: adminadmin