diff --git a/demos/demos-v2.yaml b/demos/demos-v2.yaml index 5360da21..e68c301d 100644 --- a/demos/demos-v2.yaml +++ b/demos/demos-v2.yaml @@ -313,3 +313,21 @@ demos: - name: branch 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 + documentation: https://docs.stackable.tech/home/stable/demos/openlineage-trino-airflow + stackableStack: openlineage + labels: + - trino + - airflow + - openlineage + - marquez + - data-lineage + 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 diff --git a/demos/openlineage-trino-airflow/enable-and-run-dag.yaml b/demos/openlineage-trino-airflow/enable-and-run-dag.yaml new file mode 100644 index 00000000..3e5ac10f --- /dev/null +++ b/demos/openlineage-trino-airflow/enable-and-run-dag.yaml @@ -0,0 +1,35 @@ +--- +# Waits for Airflow to be ready, then un-pauses and triggers the `openlineage_demo` DAG via the +# Airflow REST API. Running the DAG is what makes Trino execute the queries, so this is what +# populates Marquez with lineage. +apiVersion: batch/v1 +kind: Job +metadata: + name: enable-and-run-openlineage-dag +spec: + template: + spec: + containers: + - name: enable-and-run-openlineage-dag + image: oci.stackable.tech/sdp/tools:1.0.0-stackable0.0.0-dev + command: + - bash + - -euo + - pipefail + - -c + - | + kubectl rollout status --watch statefulset/airflow-webserver-default + kubectl rollout status --watch statefulset/airflow-scheduler-default + AIRFLOW_ADMIN_PASSWORD=$(cat /airflow-credentials/adminUser.password) + ACCESS_TOKEN=$(curl -XPOST http://airflow-webserver-default-headless:8080/auth/token -H 'Content-Type: application/json' -d '{"username": "admin", "password": "'$AIRFLOW_ADMIN_PASSWORD'"}' | jq -r .access_token) + curl -H "Authorization: Bearer $ACCESS_TOKEN" -H 'Content-Type: application/json' -XPATCH http://airflow-webserver-default-headless:8080/api/v2/dags/openlineage_demo -d '{"is_paused": false}' | jq + curl -H "Authorization: Bearer $ACCESS_TOKEN" -H 'Content-Type: application/json' -XPOST http://airflow-webserver-default-headless:8080/api/v2/dags/openlineage_demo/dagRuns -d '{"logical_date": null}' | jq + volumeMounts: + - name: airflow-credentials + mountPath: /airflow-credentials + volumes: + - name: airflow-credentials + secret: + secretName: airflow-credentials + restartPolicy: OnFailure + backoffLimit: 20 # give some time for the Airflow cluster to be available diff --git a/demos/openlineage-trino-airflow/rbac.yaml b/demos/openlineage-trino-airflow/rbac.yaml new file mode 100644 index 00000000..f29276c9 --- /dev/null +++ b/demos/openlineage-trino-airflow/rbac.yaml @@ -0,0 +1,37 @@ +--- +# 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`. +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: openlineage-demo-clusterrole +rules: + - apiGroups: + - "" + resources: + - pods + verbs: + - get + - list + - watch + - apiGroups: + - apps + resources: + - statefulsets + verbs: + - get + - list + - watch +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: openlineage-demo-clusterrolebinding +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: openlineage-demo-clusterrole +subjects: + - apiGroup: rbac.authorization.k8s.io + kind: Group + name: system:serviceaccounts diff --git a/docs/modules/demos/images/openlineage/airflow.png b/docs/modules/demos/images/openlineage/airflow.png new file mode 100644 index 00000000..5e6faf53 Binary files /dev/null and b/docs/modules/demos/images/openlineage/airflow.png differ diff --git a/docs/modules/demos/images/openlineage/customer-orders-2.png b/docs/modules/demos/images/openlineage/customer-orders-2.png new file mode 100644 index 00000000..f9f1e07b Binary files /dev/null and b/docs/modules/demos/images/openlineage/customer-orders-2.png differ diff --git a/docs/modules/demos/images/openlineage/job-details-customer-orders-2.png b/docs/modules/demos/images/openlineage/job-details-customer-orders-2.png new file mode 100644 index 00000000..73cfd538 Binary files /dev/null and b/docs/modules/demos/images/openlineage/job-details-customer-orders-2.png differ diff --git a/docs/modules/demos/images/openlineage/marquez.png b/docs/modules/demos/images/openlineage/marquez.png new file mode 100644 index 00000000..5298a6f7 Binary files /dev/null and b/docs/modules/demos/images/openlineage/marquez.png differ diff --git a/docs/modules/demos/images/openlineage/trino-queries.png b/docs/modules/demos/images/openlineage/trino-queries.png new file mode 100644 index 00000000..941436f3 Binary files /dev/null and b/docs/modules/demos/images/openlineage/trino-queries.png differ diff --git a/docs/modules/demos/pages/openlineage-trino-airflow.adoc b/docs/modules/demos/pages/openlineage-trino-airflow.adoc new file mode 100644 index 00000000..c1be3c69 --- /dev/null +++ b/docs/modules/demos/pages/openlineage-trino-airflow.adoc @@ -0,0 +1,217 @@ += openlineage-trino-airflow +:description: Demo showcasing OpenLineage data lineage emitted by Trino and Airflow 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 + +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. + +Install this demo on an existing Kubernetes cluster: + +[NOTE] +==== +The namespace `openlineage-demo` will be assumed in this guide. +It will be created if it doesn't exist. +==== + +[source,console] +---- +$ stackablectl demo install openlineage-trino-airflow -n openlineage-demo +---- + +[WARNING] +==== +This demo should not be run alongside other demos. +==== + +[#system-requirements] +== System requirements + +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 + +== Overview + +This demo will + +* Install the required Stackable operators +* Spin up the following products +** *Trino*: A fast distributed SQL query engine. It is configured with the built-in, first-party `openlineage` event + 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. +** *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 +* 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: + +[source] +---- +tpch.tiny.customer ─┐ + ├─> memory.demo.customer_orders +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). + +== How OpenLineage is enabled + +This demo deliberately uses only the existing SDP override mechanisms - there are *no operator or CRD changes*. This +mirrors how a future "single toggle" feature would behave, while validating the approach today. + +*Trino* loads the bundled `openlineage` event listener through a small properties file. A `ConfigMap` supplies the +file, `podOverrides` mounts it, and `configOverrides` registers it in `config.properties`: + +[source,properties] +---- +event-listener.name=openlineage +openlineage-event-listener.transport.type=HTTP +openlineage-event-listener.transport.url=http://marquez:5000 +openlineage-event-listener.transport.endpoint=/api/v1/lineage +openlineage-event-listener.trino.uri=https://trino-coordinator:8443 +openlineage-event-listener.namespace=trino-openlineage-demo +---- + +`trino.uri` is Trino's own serving address; it names the *dataset* namespace, so pointing it at the same +`trino-coordinator:8443` that Airflow uses lets both producers report the same tables under one namespace. +`namespace` sets an explicit, colon-free *job* namespace - without it Marquez cannot parse the node ID of a +Trino query job (its ID starts with a digit). See the {trino-openlineage-docs}[Trino OpenLineage event +listener docs{external-link-icon}^] for details. + +*Airflow* only needs the provider pointed at Marquez, which is done with `envOverrides`: + +[source] +---- +AIRFLOW__OPENLINEAGE__TRANSPORT='{"type": "http", "url": "http://marquez:5000", "endpoint": "api/v1/lineage"}' +AIRFLOW__OPENLINEAGE__NAMESPACE="airflow-openlineage-demo" +---- + +See the {airflow-openlineage-docs}[Airflow OpenLineage provider docs{external-link-icon}^] for details. + +== List deployed Stackable services + +To list the installed Stackable services run the following command: + +[source,console] +---- +$ stackablectl stacklet list --namespace openlineage_demo + +┌─────────┬─────────┬───────────────────┬──────────────────────────────────────────┬─────────────────────────────────┐ +│ PRODUCT ┆ NAME ┆ NAMESPACE ┆ ENDPOINTS ┆ CONDITIONS │ +╞═════════╪═════════╪═══════════════════╪══════════════════════════════════════════╪═════════════════════════════════╡ +│ airflow ┆ airflow ┆ openlineage-demo ┆ webserver-http http://172.19.0.3:31483 ┆ Available, Reconciling, Running │ +├╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┤ +│ trino ┆ trino ┆ openlineage-demo ┆ coordinator-https https://172.19.0.5:31087 ┆ Available, Reconciling, Running │ +└─────────┴─────────┴───────────────────┴──────────────────────────────────────────┴─────────────────────────────────┘ +---- + +include::partial$instance-hint.adoc[] + +[NOTE] +==== +Marquez is not an SDP product, so it does not appear in `stackablectl stacklet list`. +==== + +== Verify the demo worked + +The demo automatically triggers the `openlineage_demo` DAG once during installation, so a completed run +and its lineage should already be present. You can confirm the full flow - Airflow orchestrating, Trino +executing, and Marquez collecting the lineage - across three UIs. + +=== Airflow: the DAG tasks ran + +Open the Airflow web UI using the `webserver-http` endpoint from the `stackablectl stacklet list` output +above, and log in with the username `admin` and the password `adminadmin`. + +Under *Dags* -> `openlineage_demo` -> *Task Instances* you should see the four tasks of the pipeline - +`create_schema`, `load_customers`, `load_orders` and `customer_orders` - all in the `success` state: + +image::openlineage/airflow.png[Airflow Task Instances view showing the four openlineage_demo tasks in the success state] + +Each task is a `SQLExecuteQueryOperator` that submits a statement to Trino. + +=== Trino: the queries finished + +Open the Trino web UI using the `coordinator-https` endpoint from the `stackablectl stacklet list` +output, and log in with the username `admin` and the password `adminadmin`. + +The *Query Details* page lists the statements Airflow submitted - the `CREATE SCHEMA`, `DROP TABLE` and +`CREATE TABLE ... AS SELECT` statements that build the `memory.demo.*` tables from the `tpch.tiny.*` +source - all in the `FINISHED` state. Their *Source* is `airflow`, confirming they originated from the +DAG rather than an interactive session: + +image::openlineage/trino-queries.png[Trino Query Details page showing the finished CREATE TABLE queries submitted by Airflow] + +=== Marquez: the lineage was collected + +Marquez does not have an external listener. Forward its web UI to your machine: + +[source,console] +---- +$ kubectl -n openlineage-demo port-forward svc/marquez-web 3000:3000 +---- + +Then open http://localhost:3000 in your browser. + +The landing page lists the OpenLineage *Jobs* that Trino and Airflow reported, together with the event, +dataset and source counts. You should see the `openlineage_demo` DAG job and the per-query jobs such as +`openlineage_demo.customer_orders.query.*`, all *COMPLETED*: + +image::openlineage/marquez.png[Marquez home page listing the collected OpenLineage jobs as COMPLETED] + +Selecting a job opens its details. For example, `openlineage_demo.load_orders.query.2` shows the exact +`CREATE TABLE memory.demo.orders AS SELECT ...` statement along with its OpenLineage *job* and *run* +facets rendered as JSON - including the `sql` facet, the `AIRFLOW` integration, and the `externalQuery` +facet that links back to the originating Trino query ID: + +image::openlineage/job-details-customer-orders-2.png[Marquez job details for load_orders showing the SQL statement and OpenLineage facets as JSON] + +== Explore the lineage in Marquez + +With the Marquez UI still open at http://localhost:3000, use the namespace selector at the top to switch +between the namespaces that produced the lineage: + +* `trino-openlineage-demo` - the Trino query jobs (one per executed statement), set via + `openlineage-event-listener.namespace`, and +* `trino://trino-coordinator:8443` - the datasets (`tpch.tiny.*` and `memory.demo.*`), named after the Trino + coordinator's address. Both the Trino event listener and the Airflow SQL operator report the tables under + this shared namespace, and +* `airflow-openlineage-demo` - the `openlineage_demo` DAG and its tasks. + +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: + +image::openlineage/customer-orders-2.png[Marquez lineage graph tracing memory.demo.customer_orders upstream through the jobs to tpch.tiny.customer and tpch.tiny.orders] + +[NOTE] +==== +Trino and Airflow report lineage independently but, because `trino.uri` matches the address Airflow uses for its +Trino connection, both name the tables under the same `trino://trino-coordinator:8443` namespace. Their lineage +therefore converges on one shared set of dataset nodes: Trino contributes column-level detail per query, while +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. + +== 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. +* Read more about {openlineage}[OpenLineage{external-link-icon}^] and {marquez}[Marquez{external-link-icon}^]. diff --git a/docs/modules/demos/partials/demos.adoc b/docs/modules/demos/partials/demos.adoc index 766d2f6c..5a6d162e 100644 --- a/docs/modules/demos/partials/demos.adoc +++ b/docs/modules/demos/partials/demos.adoc @@ -8,6 +8,7 @@ * xref:logging.adoc[] * xref:nifi-kafka-druid-earthquake-data.adoc[] * xref:nifi-kafka-druid-water-level-data.adoc[] +* xref:openlineage-trino-airflow.adoc[] * xref:opensearch-rag.adoc[] * xref:signal-processing.adoc[] * xref:spark-k8s-anomaly-detection-taxi-data.adoc[] diff --git a/stacks/openlineage/airflow.yaml b/stacks/openlineage/airflow.yaml new file mode 100644 index 00000000..ede5c2ed --- /dev/null +++ b/stacks/openlineage/airflow.yaml @@ -0,0 +1,188 @@ +# Airflow with the native OpenLineage provider enabled. +# +# As with Trino, OpenLineage is switched on here *without* any operator/CRD changes - purely via +# `envOverrides`. The Airflow OpenLineage provider (`apache-airflow-providers-openlineage`) activates +# automatically once `AIRFLOW__OPENLINEAGE__TRANSPORT` is set. Because the demo DAG uses the Trino +# provider's SQL operator, Airflow also reports dataset-level lineage, which shows up in Marquez next +# to the query lineage emitted by Trino itself. +# +# NOTE: this file is intentionally NOT wrapped in a Jinja raw block (unlike other airflow stacks) +# so that the NAMESPACE / password placeholders are templated by Tera. Keep this whole file - even +# comments - free of literal Tera tag sequences (double curly braces and percent-brace tags): Tera +# processes the raw bytes before YAML parsing, so a tag token in a comment is still interpreted. +# (An earlier version of this comment quoted a raw-block tag here, which opened an unclosed raw +# block and broke rendering with "expected tag" at EOF.) +--- +apiVersion: airflow.stackable.tech/v1alpha2 +kind: AirflowCluster +metadata: + name: airflow +spec: + image: + productVersion: 3.2.2 + pullPolicy: IfNotPresent + clusterConfig: + loadExamples: false + credentialsSecretName: airflow-credentials + # The airflow-operator 'dev' schema moved the metadata DB connection out of the credentials + # Secret into its own block. Points at the postgresql-airflow chart deployed by this stack. + metadataDatabase: + postgresql: + host: postgresql-airflow + database: airflow + credentialsSecretName: airflow-postgresql-credentials + volumes: + - name: airflow-dags + configMap: + name: airflow-dags + volumeMounts: + - name: airflow-dags + mountPath: /dags/openlineage_demo.py + subPath: openlineage_demo.py + webservers: + roleConfig: + listenerClass: external-stable + config: + resources: + cpu: + min: "2" + max: "3" + memory: + limit: 3Gi + envOverrides: &envOverrides + AIRFLOW__CORE__DAGS_FOLDER: "/dags" + # Point the OpenLineage provider at Marquez. This single block is the whole "toggle". + AIRFLOW__OPENLINEAGE__TRANSPORT: '{"type": "http", "url": "http://marquez:5000", "endpoint": "api/v1/lineage"}' + AIRFLOW__OPENLINEAGE__NAMESPACE: "airflow-openlineage-demo" + # 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}}' + configOverrides: + webserver_config.py: + # Allow "POST /login/" without CSRF token + WTF_CSRF_ENABLED: "False" + roleGroups: + default: + replicas: 1 + kubernetesExecutors: + # Task pods must carry the same env so the OpenLineage listener fires and the Trino + # connection is available when the SQL operator runs. + envOverrides: *envOverrides + schedulers: + envOverrides: *envOverrides + roleGroups: + default: + replicas: 1 + dagProcessors: + envOverrides: *envOverrides + roleGroups: + default: + replicas: 1 + triggerers: + envOverrides: *envOverrides + roleGroups: + default: + replicas: 1 +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: airflow-dags +data: + openlineage_demo.py: | + """ + A minimal ELT pipeline against Trino that produces clean, layered lineage: + + tpch.tiny.customer ─┐ + ├─> memory.demo.customer_orders + 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. + """ + from datetime import datetime + + from airflow import DAG + from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator + + with DAG( + dag_id="openlineage_demo", + schedule=None, + start_date=datetime(2024, 1, 1), + catchup=False, + tags=["openlineage", "trino"], + ) as dag: + create_schema = SQLExecuteQueryOperator( + task_id="create_schema", + conn_id="trino_default", + sql="CREATE SCHEMA IF NOT EXISTS memory.demo", + ) + + 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" + ), + ], + ) + + 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" + ), + ], + ) + + 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" + ), + ], + ) + + create_schema >> [load_customers, load_orders] >> customer_orders +--- +apiVersion: v1 +kind: Secret +metadata: + name: airflow-credentials +type: Opaque +stringData: + adminUser.username: admin + adminUser.firstname: Airflow + adminUser.lastname: Admin + adminUser.email: airflow@airflow.com + adminUser.password: "{{ airflowAdminPassword }}" +--- +# Credentials for the Airflow metadata database (the postgresql-airflow chart in this stack). +# The operator reads the `username`/`password` keys and builds the SQLAlchemy URL from the +# clusterConfig.metadataDatabase block above. +apiVersion: v1 +kind: Secret +metadata: + name: airflow-postgresql-credentials +type: Opaque +stringData: + username: airflow + password: airflow diff --git a/stacks/openlineage/marquez.yaml b/stacks/openlineage/marquez.yaml new file mode 100644 index 00000000..5e211ab9 --- /dev/null +++ b/stacks/openlineage/marquez.yaml @@ -0,0 +1,546 @@ +# Marquez is the OpenLineage reference metadata server + web UI, used here as the consuming +# application for the lineage that Trino and Airflow emit. It is only part of this *demo* - it is +# not an SDP product. +# +# VENDORED MANIFESTS - do not hand-edit the resources below; regenerate instead. +# Marquez publishes NO Helm repository or OCI chart (the old +# https://marquezproject.github.io/marquez/ repo URL 404s), so we render the upstream chart to +# static YAML and ship that as a plainYaml manifest. +# +# Chart: marquez 0.51.1 (Apache-2.0) - https://github.com/MarquezProject/marquez/tree/main/chart +# Regenerate: +# git clone --depth 1 --branch 0.51.1 https://github.com/MarquezProject/marquez.git +# helm repo add bitnami https://charts.bitnami.com/bitnami # for the postgresql/common subcharts +# ( cd marquez/chart && helm dependency build ) +# helm template marquez ./marquez/chart -f VALUES --skip-tests \ +# | grep -v '^ namespace: "default"$' > marquez.yaml # strip test hooks + hardcoded ns +# # ...then re-add this header. +# +# VALUES used: +# fullnameOverride: marquez +# service: {type: ClusterIP, port: 5000} # API Service marquez:5000 (Trino/Airflow target it and +# # the web pod calls it via MARQUEZ_PORT, all internal) +# web.enabled: true +# postgresql: {enabled: true, auth: marquez/marquez/marquez, +# image.repository: bitnamilegacy/postgresql, global...allowInsecureImages: true} +# +# POST-RENDER EDITS (the chart forces ONE shared service.{type,port} onto both the API and web +# Services; there is no per-service override, so these two Service specs are hand-adjusted after +# rendering - see the "POST-RENDER EDIT" markers below): +# * marquez (API): kept ClusterIP :5000 - consumed only cluster-internally. +# * marquez-web (UI) : set to NodePort :3000 - the only component end users reach; port now +# matches its container (chart would otherwise publish the UI on :5000). +# +# Resulting Services: marquez:5000 (API, ClusterIP), marquez-web:3000 (UI, NodePort), +# marquez-postgresql:5432 (ClusterIP). +--- +# Source: marquez/charts/postgresql/templates/secrets.yaml +apiVersion: v1 +kind: Secret +metadata: + name: marquez-postgresql + labels: + app.kubernetes.io/name: postgresql + helm.sh/chart: postgresql-11.8.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm +type: Opaque +data: + postgres-password: "b1o2ZjI2OGFWSA==" + password: "bWFycXVleg==" + # We don't auto-generate LDAP password when it's not provided as we do for other passwords +--- +# Source: marquez/templates/marquez/secret.yaml +apiVersion: v1 +kind: Secret +metadata: + name: marquez + labels: + app.kubernetes.io/name: marquez + helm.sh/chart: marquez-0.51.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm +type: Opaque +data: + marquez-db-password: "bWFjb25kbw==" + + +--- +# Source: marquez/templates/marquez/configmap.yaml +apiVersion: v1 +kind: ConfigMap +metadata: + name: marquez-config + labels: + app.kubernetes.io/name: marquez + helm.sh/chart: marquez-0.51.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm + app.kubernetes.io/component: marquez + annotations: +data: + ### HTTP SERVER CONFIG ### + + config.yml: |- + # Enables HTTP server configuration overrides (see: https://www.dropwizard.io/en/stable/manual/configuration.html#servers) + server: + applicationConnectors: + - type: http + port: ${MARQUEZ_PORT} + httpCompliance: RFC7230_LEGACY + adminConnectors: + - type: http + port: ${MARQUEZ_ADMIN_PORT} + + ### DATABASE CONFIG ### + + # Enables database configuration overrides (see: https://www.dropwizard.io/en/stable/manual/configuration.html#database) + db: + driverClass: org.postgresql.Driver + url: jdbc:postgresql://${POSTGRES_HOST}:${POSTGRES_PORT}/${POSTGRES_DB} + user: ${POSTGRES_USER} + password: ${POSTGRES_PASSWORD} + # Enables retention policy configuration for database (default: disabled) + dbRetention: + frequencyMins: 15 + numberOfRowsPerBatch: 1000 + retentionDays: 7 + + # Enables database migration on startup (default: true) + migrateOnStartup: ${MIGRATE_ON_STARTUP} + logging: + level: ${LOG_LEVEL} + appenders: + - type: console + +--- +# Source: marquez/charts/postgresql/templates/primary/svc-headless.yaml +apiVersion: v1 +kind: Service +metadata: + name: marquez-postgresql-hl + labels: + app.kubernetes.io/name: postgresql + helm.sh/chart: postgresql-11.8.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm + app.kubernetes.io/component: primary + # Use this annotation in addition to the actual publishNotReadyAddresses + # field below because the annotation will stop being respected soon but the + # field is broken in some versions of Kubernetes: + # https://github.com/kubernetes/kubernetes/issues/58662 + service.alpha.kubernetes.io/tolerate-unready-endpoints: "true" +spec: + type: ClusterIP + clusterIP: None + # We want all pods in the StatefulSet to have their addresses published for + # the sake of the other Postgresql pods even before they're ready, since they + # have to be able to talk to each other in order to become ready. + publishNotReadyAddresses: true + ports: + - name: tcp-postgresql + port: 5432 + targetPort: tcp-postgresql + selector: + app.kubernetes.io/name: postgresql + app.kubernetes.io/instance: marquez + app.kubernetes.io/component: primary + +--- +# Source: marquez/charts/postgresql/templates/primary/svc.yaml +apiVersion: v1 +kind: Service +metadata: + name: marquez-postgresql + labels: + app.kubernetes.io/name: postgresql + helm.sh/chart: postgresql-11.8.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm + app.kubernetes.io/component: primary + annotations: +spec: + type: ClusterIP + sessionAffinity: None + ports: + - name: tcp-postgresql + port: 5432 + targetPort: tcp-postgresql + nodePort: null + selector: + app.kubernetes.io/name: postgresql + app.kubernetes.io/instance: marquez + app.kubernetes.io/component: primary + +--- +# Source: marquez/templates/marquez/service.yaml +apiVersion: v1 +kind: Service +metadata: + name: marquez + labels: + app.kubernetes.io/name: marquez + helm.sh/chart: marquez-0.51.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm + app.kubernetes.io/component: marquez + annotations: +spec: + # POST-RENDER EDIT: chart default is a single service.type for both Services. The API is only + # consumed cluster-internally (Trino, Airflow, and the marquez-web pod), so keep it ClusterIP. + type: ClusterIP + ports: + - port: 5000 + targetPort: http + protocol: TCP + name: http + + selector: + app.kubernetes.io/name: marquez + app.kubernetes.io/instance: marquez + app.kubernetes.io/component: marquez + +--- +# Source: marquez/templates/web/service.yaml +apiVersion: v1 +kind: Service +metadata: + name: marquez-web + labels: + app.kubernetes.io/name: marquez + helm.sh/chart: marquez-0.51.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm + app.kubernetes.io/component: web + annotations: +spec: + # POST-RENDER EDIT: the web UI is the only component end users reach, so expose it (NodePort) on + # its own container port 3000 - not the chart's shared service.port of 5000, which was confusing. + type: NodePort + ports: + - port: 3000 + targetPort: http + protocol: TCP + name: http + + selector: + app.kubernetes.io/name: marquez + app.kubernetes.io/instance: marquez + app.kubernetes.io/component: web + +--- +# Source: marquez/templates/marquez/deployment.yaml +apiVersion: apps/v1 +kind: Deployment +metadata: + name: marquez + labels: + app.kubernetes.io/name: marquez + helm.sh/chart: marquez-0.51.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm + app.kubernetes.io/component: marquez + annotations: +spec: + selector: + matchLabels: + app.kubernetes.io/name: marquez + app.kubernetes.io/instance: marquez + app.kubernetes.io/component: marquez + replicas: 1 + template: + metadata: + labels: + app.kubernetes.io/name: marquez + helm.sh/chart: marquez-0.51.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm + app.kubernetes.io/component: marquez + spec: + serviceAccountName: default + # This init container is for avoiding CrashLoopback errors in the Marquez container because the PostgreSQL container is not ready + initContainers: + - name: wait-for-db + image: docker.io/bitnamilegacy/postgresql:12.1.0 + imagePullPolicy: IfNotPresent + command: + - /bin/bash + args: + - -ec + - until pg_isready -h ${POSTGRES_HOST} -p ${POSTGRES_PORT} -U ${POSTGRES_USER}; do echo waiting for database; sleep 2; done; + env: + - name: POSTGRES_HOST + value: "marquez-postgresql" + - name: POSTGRES_PORT + value: "5432" + - name: POSTGRES_USER + value: "marquez" + containers: + - name: marquez + image: docker.io/marquezproject/marquez:0.51.1 + imagePullPolicy: IfNotPresent + volumeMounts: + - name: marquez-volume + mountPath: /usr/src/app/config.yml + subPath: config.yml + ports: + - name: http + containerPort: 5000 + protocol: TCP + - name: http-admin + containerPort: 5001 + protocol: TCP + livenessProbe: + httpGet: + path: /healthcheck + port: http-admin + readinessProbe: + httpGet: + path: /healthcheck + port: http-admin + env: + - name: MARQUEZ_PORT + value: "5000" + - name: MARQUEZ_ADMIN_PORT + value: "5001" + - name: MARQUEZ_CONFIG + value: /usr/src/app/config.yml + - name: POSTGRES_HOST + value: "marquez-postgresql" + - name: POSTGRES_PORT + value: "5432" + - name: POSTGRES_DB + value: "marquez" + - name: POSTGRES_USER + value: "marquez" + - name: POSTGRES_PASSWORD + valueFrom: + secretKeyRef: + name: marquez-postgresql + key: password + - name: MIGRATE_ON_STARTUP + value: "true" + resources: + limits: {} + requests: {} + volumes: + - name: marquez-volume + configMap: + name: marquez-config + +--- +# Source: marquez/templates/web/deployment.yaml +apiVersion: apps/v1 +kind: Deployment +metadata: + name: marquez-web + labels: + app.kubernetes.io/name: marquez + helm.sh/chart: marquez-0.51.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm + app.kubernetes.io/component: web +spec: + selector: + matchLabels: + app.kubernetes.io/name: marquez + app.kubernetes.io/instance: marquez + app.kubernetes.io/component: web + replicas: 1 + template: + metadata: + labels: + app.kubernetes.io/name: marquez + helm.sh/chart: marquez-0.51.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm + app.kubernetes.io/component: web + spec: + containers: + - name: marquez-web + image: docker.io/marquezproject/marquez-web:0.51.1 + imagePullPolicy: IfNotPresent + ports: + - name: http + containerPort: 3000 + protocol: TCP + livenessProbe: + httpGet: + path: /healthcheck + port: http + readinessProbe: + httpGet: + path: /healthcheck + port: http + env: + - name: MARQUEZ_HOST + value: marquez + - name: MARQUEZ_PORT + value: "5000" + - name: WEB_PORT + value: "3000" + resources: + limits: {} + requests: {} + +--- +# Source: marquez/charts/postgresql/templates/primary/statefulset.yaml +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: marquez-postgresql + labels: + app.kubernetes.io/name: postgresql + helm.sh/chart: postgresql-11.8.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm + app.kubernetes.io/component: primary + annotations: +spec: + replicas: 1 + serviceName: marquez-postgresql-hl + updateStrategy: + rollingUpdate: {} + type: RollingUpdate + selector: + matchLabels: + app.kubernetes.io/name: postgresql + app.kubernetes.io/instance: marquez + app.kubernetes.io/component: primary + template: + metadata: + name: marquez-postgresql + labels: + app.kubernetes.io/name: postgresql + helm.sh/chart: postgresql-11.8.1 + app.kubernetes.io/instance: marquez + app.kubernetes.io/managed-by: Helm + app.kubernetes.io/component: primary + annotations: + spec: + serviceAccountName: default + + affinity: + podAffinity: + + podAntiAffinity: + preferredDuringSchedulingIgnoredDuringExecution: + - podAffinityTerm: + labelSelector: + matchLabels: + app.kubernetes.io/name: postgresql + app.kubernetes.io/instance: marquez + app.kubernetes.io/component: primary + topologyKey: kubernetes.io/hostname + weight: 1 + nodeAffinity: + + securityContext: + fsGroup: 1001 + hostNetwork: false + hostIPC: false + initContainers: + containers: + - name: postgresql + image: docker.io/bitnamilegacy/postgresql:12.1.0 + imagePullPolicy: "IfNotPresent" + securityContext: + runAsUser: 1001 + env: + - name: BITNAMI_DEBUG + value: "false" + - name: POSTGRESQL_PORT_NUMBER + value: "5432" + - name: POSTGRESQL_VOLUME_DIR + value: "/bitnami/postgresql" + - name: PGDATA + value: "/bitnami/postgresql/data" + # Authentication + - name: POSTGRES_USER + value: "marquez" + - name: POSTGRES_POSTGRES_PASSWORD + valueFrom: + secretKeyRef: + name: marquez-postgresql + key: postgres-password + - name: POSTGRES_PASSWORD + valueFrom: + secretKeyRef: + name: marquez-postgresql + key: password + - name: POSTGRES_DB + value: "marquez" + # Replication + # Initdb + # Standby + # LDAP + - name: POSTGRESQL_ENABLE_LDAP + value: "no" + # TLS + - name: POSTGRESQL_ENABLE_TLS + value: "no" + # Audit + - name: POSTGRESQL_LOG_HOSTNAME + value: "false" + - name: POSTGRESQL_LOG_CONNECTIONS + value: "false" + - name: POSTGRESQL_LOG_DISCONNECTIONS + value: "false" + - name: POSTGRESQL_PGAUDIT_LOG_CATALOG + value: "off" + # Others + - name: POSTGRESQL_CLIENT_MIN_MESSAGES + value: "error" + - name: POSTGRESQL_SHARED_PRELOAD_LIBRARIES + value: "pgaudit" + ports: + - name: tcp-postgresql + containerPort: 5432 + livenessProbe: + failureThreshold: 6 + initialDelaySeconds: 30 + periodSeconds: 10 + successThreshold: 1 + timeoutSeconds: 5 + exec: + command: + - /bin/sh + - -c + - exec pg_isready -U "marquez" -d "dbname=marquez" -h 127.0.0.1 -p 5432 + readinessProbe: + failureThreshold: 6 + initialDelaySeconds: 5 + periodSeconds: 10 + successThreshold: 1 + timeoutSeconds: 5 + exec: + command: + - /bin/sh + - -c + - -e + + - | + exec pg_isready -U "marquez" -d "dbname=marquez" -h 127.0.0.1 -p 5432 + resources: + limits: {} + requests: + cpu: 250m + memory: 256Mi + volumeMounts: + - name: dshm + mountPath: /dev/shm + - name: data + mountPath: /bitnami/postgresql + volumes: + - name: dshm + emptyDir: + medium: Memory + volumeClaimTemplates: + - metadata: + name: data + spec: + accessModes: + - "ReadWriteOnce" + resources: + requests: + storage: "8Gi" diff --git a/stacks/openlineage/trino.yaml b/stacks/openlineage/trino.yaml new file mode 100644 index 00000000..ec779b91 --- /dev/null +++ b/stacks/openlineage/trino.yaml @@ -0,0 +1,131 @@ +# Trino with the bundled, first-party OpenLineage event listener enabled. +# +# This demo deliberately enables OpenLineage *without* any operator/CRD changes, using only the +# existing override mechanisms: +# * `configOverrides` adds `event-listener.config-files` to `config.properties`, and +# * `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. +# +# For now, on K8s 1.35, TrinoCatalogs need to be deployed before the TrinoCluster. +# See: https://github.com/stackabletech/trino-operator/issues/854 +--- +apiVersion: trino.stackable.tech/v1alpha1 +kind: TrinoCatalog +metadata: + name: tpch + labels: + trino: trino +spec: + connector: + tpch: {} +--- +apiVersion: trino.stackable.tech/v1alpha1 +kind: TrinoCatalog +metadata: + name: memory + labels: + trino: trino +spec: + connector: + generic: + connectorName: memory +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: trino-openlineage +data: + # Trino reads this file because `config.properties` points at it via `event-listener.config-files` + # (see the `configOverrides` on the coordinator below). + event-listener.properties: | + event-listener.name=openlineage + openlineage-event-listener.transport.type=HTTP + openlineage-event-listener.transport.url=http://marquez:5000 + openlineage-event-listener.transport.endpoint=/api/v1/lineage + # Must be Trino's real serving address (HTTPS on 8443 - TLS is enabled and 8080 is not exposed). + # This is the naming string for the *dataset* namespace; matching the address Airflow uses in its + # Trino connection makes both producers name the same tables `trino://trino-coordinator:8443`, so + # their lineage attaches to one shared set of dataset nodes instead of two duplicate namespaces. + openlineage-event-listener.trino.uri=https://trino-coordinator:8443 + # Set an explicit, colon-free job namespace. Without this the namespace defaults to the dataset + # namespace (`trino://trino-coordinator:8443`), and Marquez's NodeId parser cannot split + # `job:trino://trino-coordinator:8443:` because Trino query IDs start with a digit - + # its URI heuristic then treats the `:` delimiter as a `host:port` colon and 500s when + # a job is opened. A plain namespace keeps job NodeIds parseable. + openlineage-event-listener.namespace=trino-openlineage-demo +--- +apiVersion: trino.stackable.tech/v1alpha1 +kind: TrinoCluster +metadata: + name: trino +spec: + image: + productVersion: "481" + clusterConfig: + catalogLabelSelector: + matchLabels: + trino: trino + authentication: + - authenticationClass: trino-users + tls: + # Static (password) authentication requires TLS. `tls` is the SecretClass that + # secret-operator ships out of the box, so nothing extra needs to be defined here. + serverSecretClass: tls + coordinators: + roleConfig: + listenerClass: external-stable + config: + resources: + cpu: + min: "500m" + max: "2" + memory: + limit: "4Gi" + # The event listener only needs to run on the coordinator: that is where query lifecycle + # events (and their table-/column-level lineage) are produced. + configOverrides: + config.properties: + event-listener.config-files: /stackable/mount/openlineage/event-listener.properties + podOverrides: + spec: + containers: + - name: trino + volumeMounts: + - name: openlineage-config + mountPath: /stackable/mount/openlineage + volumes: + - name: openlineage-config + configMap: + name: trino-openlineage + roleGroups: + default: + replicas: 1 + workers: + config: + resources: + cpu: + min: "1" + max: "4" + memory: + limit: "4Gi" + roleGroups: + default: + replicas: 1 +--- +apiVersion: authentication.stackable.tech/v1alpha1 +kind: AuthenticationClass +metadata: + name: trino-users +spec: + provider: + static: + userCredentialsSecret: + name: trino-users +--- +apiVersion: v1 +kind: Secret +metadata: + name: trino-users +type: kubernetes.io/opaque +stringData: + admin: "{{ trinoAdminPassword }}" diff --git a/stacks/stacks-v2.yaml b/stacks/stacks-v2.yaml index 43ef0fd3..09ee5859 100644 --- a/stacks/stacks-v2.yaml +++ b/stacks/stacks-v2.yaml @@ -683,3 +683,38 @@ stacks: - name: branch description: Git branch to download demo files from default: main + openlineage: + description: >- + Stack for the OpenLineage demo: Trino and Airflow both emitting OpenLineage events to Marquez + stackableRelease: dev + stackableOperators: + - commons + - listener + - secret + - trino + - airflow + labels: + - trino + - airflow + - openlineage + - marquez + - data-lineage + manifests: + - helmChart: stacks/_templates/postgresql-airflow.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 + - plainYaml: stacks/openlineage/airflow.yaml + supportedNamespaces: [] + resourceRequests: + cpu: 5000m + memory: 12Gi + pvc: 15Gi + parameters: + - name: trinoAdminPassword + description: Password of the Trino admin user + default: adminadmin + - name: airflowAdminPassword + description: Password of the Airflow admin user + default: adminadmin