Skip to content
Draft
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ All notable changes to this project will be documented in this file.

- BREAKING: Add required CLI argument and env var to set the image repository used to construct final product image names: `IMAGE_REPOSITORY` (`--image-repository`), eg. `oci.example.org/my/namespace` ([#684]).
- Support for Spark `4.1.2` ([#708])
- Add `spec.openLineage` to enable OpenLineage lineage emission: injects the OpenLineage Spark listener, HTTP transport (endpoint resolved from a discovery ConfigMap), a stable job name, and the required `--add-opens` JVM flag ([#XXXX]).

### Fixed

Expand Down Expand Up @@ -44,6 +45,7 @@ All notable changes to this project will be documented in this file.
[#705]: https://github.com/stackabletech/spark-k8s-operator/pull/705
[#708]: https://github.com/stackabletech/spark-k8s-operator/pull/708
[#716]: https://github.com/stackabletech/spark-k8s-operator/pull/716
[#XXXX]: https://github.com/stackabletech/spark-k8s-operator/pull/XXXX

## [26.3.0] - 2026-03-16

Expand Down
73 changes: 73 additions & 0 deletions docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
= OpenLineage

https://openlineage.io/[OpenLineage] is an open standard for data lineage collection.
The Spark operator can automatically emit lineage events for a `SparkApplication` to an OpenLineage-compatible backend such as https://marquezproject.github.io/marquez/[Marquez], without any manual `sparkConf` wiring.

When enabled, the operator injects everything required to make the https://openlineage.io/docs/integrations/spark/[OpenLineage Spark listener] work:

* the OpenLineage Spark listener (appended to `spark.extraListeners`, so any listener you configure yourself is preserved),
* the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so it shares the same classloader as connectors pulled in through xref:usage-guide/job-dependencies.adoc[`deps.packages`] (for example Apache Iceberg),
* an HTTP transport pointing at the backend endpoint,
* a stable lineage namespace and job name (see <<job-name>>), and
* the `--add-opens java.base/java.security=ALL-UNNAMED` JVM flag the listener requires on the driver and executors.

== Enabling OpenLineage

The backend endpoint is not configured on the `SparkApplication` directly.
Instead, mirroring the xref:usage-guide/logging.adoc[vector aggregator discovery], the operator reads it from a discovery ConfigMap referenced by `configMapName`:

[source,yaml]
----
apiVersion: spark.stackable.tech/v1alpha1
kind: SparkApplication
metadata:
name: orders-by-nation
spec:
openLineage:
enabled: true # <1>
configMapName: marquez-discovery # <2>
# namespace: orders-lineage # <3>
# appName: orders-by-nation # <4>
...
----
<1> Enable OpenLineage event emission. Defaults to `false`, in which case the operator emits nothing OpenLineage-related and behaviour is unchanged.
<2> Name of a discovery ConfigMap holding the backend endpoint (see below).
Must be in the same namespace as the `SparkApplication`.
<3> Optional. The OpenLineage namespace events are reported under.
Defaults to the application's Kubernetes namespace.
<4> Optional. The stable OpenLineage job name (see <<job-name>>).

The discovery ConfigMap holds the backend base URL under the `ADDRESS` key:

.Example discovery ConfigMap
[source,yaml]
----
apiVersion: v1
kind: ConfigMap
metadata:
name: marquez-discovery
data:
ADDRESS: http://marquez:5000 # <1>
----
<1> Base URL of the OpenLineage backend, for example the Marquez API service.

Alternatively, you can set the endpoint directly through `sparkConf` (`spark.openlineage.transport.url`).
If `enabled: true` and no endpoint can be resolved from either the discovery ConfigMap or `sparkConf`, reconciliation fails with a clear error rather than emitting a config that silently drops events.

[#job-name]
== Job name

The OpenLineage job name is resolved in the following order:

. `spec.openLineage.appName`, if set.
. `spark.app.name` from `sparkConf`, if set.
. `metadata.name` as a last resort.

The last case additionally emits a Kubernetes warning event.
Falling back to `metadata.name` is discouraged: if that name carries a timestamp or other run-specific suffix (as generated names often do), every run becomes a separate job in the backend and the run history fragments.
Set `spec.openLineage.appName` (or `spark.app.name`) to a stable value to keep all runs of the same logical job together.

== Overriding injected configuration

Every injected value is a default that goes into the submit configuration *before* your xref:usage-guide/overrides.adoc[`sparkConf`] is merged, so any key you set yourself wins.
`spark.extraListeners` and `spark.jars` are the exception: the operator *appends* to any value you provide (comma-merged) rather than overwriting it, so your own listeners and jars are kept alongside OpenLineage's.
1 change: 1 addition & 0 deletions docs/modules/spark-k8s/partials/nav.adoc
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
** xref:spark-k8s:usage-guide/app_templates.adoc[]
** xref:spark-k8s:usage-guide/security.adoc[]
** xref:spark-k8s:usage-guide/logging.adoc[]
** xref:spark-k8s:usage-guide/openlineage.adoc[]
** xref:spark-k8s:usage-guide/history-server.adoc[]
** xref:spark-k8s:usage-guide/spark-connect.adoc[]
** xref:spark-k8s:usage-guide/examples.adoc[]
Expand Down
72 changes: 72 additions & 0 deletions extra/crds.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -1957,6 +1957,42 @@ spec:
- cluster
- client
type: string
openLineage:
description: |-
Emit [OpenLineage](https://openlineage.io/) lineage events for this application.
The OpenLineage Spark listener runs on the driver and describes the whole application,
so this is application-scoped config (not per driver/executor role).
See the [OpenLineage usage guide](https://docs.stackable.tech/home/nightly/spark-k8s/usage-guide/openlineage).
nullable: true
properties:
appName:
description: |-
A stable OpenLineage job/application name. Setting this prevents fragmented run history
(and the intermittent `unknown` job-name bug). If unset, the operator resolves it from
`spark.app.name`, falling back to `metadata.name` (with a warning event).
nullable: true
type: string
configMapName:
description: |-
Name of the OpenLineage backend [discovery ConfigMap](https://docs.stackable.tech/home/nightly/concepts/service_discovery).
It must contain the key `ADDRESS` with the base URL of the OpenLineage backend
(e.g. `http://marquez:5000`). Mirrors the `vectorAggregatorConfigMapName` field on this CRD.
maxLength: 253
minLength: 1
nullable: true
pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
type: string
enabled:
default: false
description: Enable OpenLineage event emission. Defaults to `false` (nothing is injected).
type: boolean
namespace:
description: |-
The OpenLineage namespace lineage is reported under.
Defaults to the application's Kubernetes namespace (`metadata.namespace`).
nullable: true
type: string
type: object
s3connection:
description: |-
Configure an S3 connection that the SparkApplication has access to.
Expand Down Expand Up @@ -6563,6 +6599,42 @@ spec:
- cluster
- client
type: string
openLineage:
description: |-
Emit [OpenLineage](https://openlineage.io/) lineage events for this application.
The OpenLineage Spark listener runs on the driver and describes the whole application,
so this is application-scoped config (not per driver/executor role).
See the [OpenLineage usage guide](https://docs.stackable.tech/home/nightly/spark-k8s/usage-guide/openlineage).
nullable: true
properties:
appName:
description: |-
A stable OpenLineage job/application name. Setting this prevents fragmented run history
(and the intermittent `unknown` job-name bug). If unset, the operator resolves it from
`spark.app.name`, falling back to `metadata.name` (with a warning event).
nullable: true
type: string
configMapName:
description: |-
Name of the OpenLineage backend [discovery ConfigMap](https://docs.stackable.tech/home/nightly/concepts/service_discovery).
It must contain the key `ADDRESS` with the base URL of the OpenLineage backend
(e.g. `http://marquez:5000`). Mirrors the `vectorAggregatorConfigMapName` field on this CRD.
maxLength: 253
minLength: 1
nullable: true
pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
type: string
enabled:
default: false
description: Enable OpenLineage event emission. Defaults to `false` (nothing is injected).
type: boolean
namespace:
description: |-
The OpenLineage namespace lineage is reported under.
Defaults to the application's Kubernetes namespace (`metadata.namespace`).
nullable: true
type: string
type: object
s3connection:
description: |-
Configure an S3 connection that the SparkApplication has access to.
Expand Down
42 changes: 40 additions & 2 deletions rust/operator-binary/src/config/jvm.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,8 @@ use stackable_operator::crd::s3;

use crate::crd::{
constants::{
JVM_SECURITY_PROPERTIES_FILE, STACKABLE_TLS_STORE_PASSWORD, STACKABLE_TRUST_STORE,
VOLUME_MOUNT_PATH_LOG_CONFIG,
JVM_SECURITY_PROPERTIES_FILE, OPENLINEAGE_ADD_OPENS, STACKABLE_TLS_STORE_PASSWORD,
STACKABLE_TRUST_STORE, VOLUME_MOUNT_PATH_LOG_CONFIG,
},
logdir::ResolvedLogDir,
tlscerts::tls_secret_names,
Expand Down Expand Up @@ -36,6 +36,18 @@ pub fn construct_extra_java_options(
]);
}

// OpenLineage on Spark 4.x needs `java.base/java.security` opened to the unnamed module,
// otherwise the driver throws a non-fatal `InaccessibleObjectException` and silently degrades
// extension-interface lineage (MVP §7). Added to both driver and executor.
if spark_application

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

see other comment, this needs proper Spark 3 treatment.

.spec
.open_lineage
.as_ref()
.is_some_and(|open_lineage| open_lineage.enabled)
{
jvm_args.push(OPENLINEAGE_ADD_OPENS.to_string());
}

// The role's `jvmArgumentOverrides` are applied on top of the operator-generated arguments
// above. Note this is not purely additive: a role may also remove or replace operator-set
// arguments (e.g. a `removeRegex` dropping the `-Djava.security.properties` default) — see the
Expand Down Expand Up @@ -132,4 +144,30 @@ mod tests {
"-Dhttps.proxyHost=from-executor"
);
}

#[test]
fn test_construct_jvm_arguments_openlineage_add_opens() {
let input = r#"
apiVersion: spark.stackable.tech/v1alpha1
kind: SparkApplication
metadata:
name: spark-example
spec:
mode: cluster
mainApplicationFile: test.py
sparkImage:
productVersion: 1.2.3
openLineage:
enabled: true
"#;

let deserializer = serde_yaml::Deserializer::from_str(input);
let spark_app: SparkApplication =
serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap();
let (driver_extra_java_options, executor_extra_java_options) =
construct_extra_java_options(&spark_app, &None, &None);

assert!(driver_extra_java_options.contains(OPENLINEAGE_ADD_OPENS));
assert!(executor_extra_java_options.contains(OPENLINEAGE_ADD_OPENS));
}
}
24 changes: 24 additions & 0 deletions rust/operator-binary/src/crd/constants.rs
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,30 @@ pub const DEFAULT_LISTENER_CLASS: &str = "cluster-internal";

pub const DEFAULT_SUBMIT_JOB_RETRY_ON_FAILURE_COUNT: u16 = 0;

// --- OpenLineage (see the OpenLineage usage guide) ---

/// The OpenLineage Spark listener, appended to `spark.extraListeners` when OpenLineage is enabled.
pub const OPENLINEAGE_LISTENER_CLASS: &str = "io.openlineage.spark.agent.OpenLineageSparkListener";

/// `local://` URI of the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so
/// it shares the (child) classloader with `--packages` connectors. Delivering it this way — rather
/// than on `extraClassPath` (system classloader) — is what lets OpenLineage's connector probes
/// succeed; see the classpath discussion in the OpenLineage usage guide.
///
/// IMPORTANT: the version MUST stay in sync with the `openlineage-spark-version` build-argument in
/// `docker-images/spark-k8s/boil-config.toml` (the image is what places the jar at this path).
pub const OPENLINEAGE_JAR_LOCAL_URI: &str =
"local:///stackable/spark/openlineage/openlineage-spark_2.13-1.51.0.jar";

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this needs a way that is version agnostic / some proper way to keep this in sync other than a comment.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The way we solved similar problems in the past was to use symlinks


/// The key the OpenLineage discovery ConfigMap must hold, carrying the backend base URL. Mirrors the
/// `ADDRESS` contract of the existing `vectorAggregatorConfigMapName` discovery ConfigMap.
pub const OPENLINEAGE_CONFIG_MAP_ADDRESS_KEY: &str = "ADDRESS";

/// Java module-system flag OpenLineage requires on Spark 4.x: without it the driver throws a
Comment thread
sweb marked this conversation as resolved.
/// non-fatal `InaccessibleObjectException` and silently degrades extension-interface lineage.
/// Appended to both driver and executor `extraJavaOptions`.
pub const OPENLINEAGE_ADD_OPENS: &str = "--add-opens java.base/java.security=ALL-UNNAMED";

/// The JVM `security.properties` entries the operator sets by default (DNS cache TTLs).
pub fn default_jvm_security_properties() -> BTreeMap<String, String> {
[
Expand Down
Loading
Loading