diff --git a/CHANGELOG.md b/CHANGELOG.md index e198c570..88bc364a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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) and a stable job name. The listener jar is referenced through a stable, version- and Scala-independent symlink maintained by the Spark image, so the operator tracks neither. The `--add-opens java.base/java.security` JVM flag is emitted only on Spark 4.x (it is unnecessary on the Spark 3.5.x images and must not be set there) ([#XXXX]). ### Fixed @@ -30,7 +31,7 @@ All notable changes to this project will be documented in this file. - Internal operator refactoring: introduce dereference() and validate() steps in the reconciler for spark application, spark connect and spark history server([#687]). - test: Bump vector-aggregator to 0.55.0, replace /graphql call with gRPC call ([#689]). - Fix the `SparkApplication` CRD description, which incorrectly described it as a "Spark cluster stacklet" rather than a Spark application ([#705]). -- BREAKING: make application templates namespaced instead of cluster wide objects ([#694]). +- BREAKING: make application templates namespaced instead of cluster wide objects ([#694]), reverted in [#719] and readded in [#720]. [#674]: https://github.com/stackabletech/spark-k8s-operator/pull/674 [#679]: https://github.com/stackabletech/spark-k8s-operator/pull/679 @@ -44,6 +45,8 @@ 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 +[#719]: https://github.com/stackabletech/spark-k8s-operator/pull/719 +[#720]: https://github.com/stackabletech/spark-k8s-operator/pull/720 ## [26.3.0] - 2026-03-16 diff --git a/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc b/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc new file mode 100644 index 00000000..e8aa52f3 --- /dev/null +++ b/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc @@ -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). The operator references a stable symlink maintained by the image, which points at the correct build for that image's Spark version (Scala 2.13 for Spark 4.x, Scala 2.12 for Spark 3.5.x), so neither the jar version nor its Scala suffix has to be tracked in the operator, +* an HTTP transport pointing at the backend endpoint, +* a stable lineage namespace and job name (see <>), and +* on Spark 4.x only, the `--add-opens java.base/java.security=ALL-UNNAMED` JVM flag the listener requires on the driver and executors. This flag is not added on the Spark 3.5.x images, where it is unnecessary (and, on their JVM, potentially harmful). + +== 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 <>). + +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. diff --git a/docs/modules/spark-k8s/partials/nav.adoc b/docs/modules/spark-k8s/partials/nav.adoc index 71f2278e..815c58dd 100644 --- a/docs/modules/spark-k8s/partials/nav.adoc +++ b/docs/modules/spark-k8s/partials/nav.adoc @@ -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[] diff --git a/docs/modules/spark-k8s/partials/supported-versions.adoc b/docs/modules/spark-k8s/partials/supported-versions.adoc index bd3cc078..494d69e6 100644 --- a/docs/modules/spark-k8s/partials/supported-versions.adoc +++ b/docs/modules/spark-k8s/partials/supported-versions.adoc @@ -5,7 +5,7 @@ - 4.1.2 (Hadoop 3.4.3, Scala 2.13, Python 3.12, Java 21) (LTS) - 4.1.1 (Hadoop 3.4.3, Scala 2.13, Python 3.12, Java 21) (Deprecated) -- 3.5.8 (Hadoop 3.4.2, Scala 2.12, Python 3.11, Java 17) (Deprecated) +- 3.5.8 (HBase 2.6.6, Hadoop 3.4.3, Scala 2.12, Python 3.11, Java 17) (Deprecated) Apache Spark 4.1.1 & 4.1.2 have the following known issues (as of July 2026): diff --git a/extra/crds.yaml b/extra/crds.yaml index ab28308f..56f60237 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -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. @@ -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. diff --git a/rust/operator-binary/src/config/jvm.rs b/rust/operator-binary/src/config/jvm.rs index cc5388b8..0c45972a 100644 --- a/rust/operator-binary/src/config/jvm.rs +++ b/rust/operator-binary/src/config/jvm.rs @@ -1,11 +1,13 @@ use stackable_operator::crd::s3; use crate::crd::{ + Error, 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, + spark_major_version, tlscerts::tls_secret_names, v1alpha1::SparkApplication, }; @@ -16,11 +18,15 @@ use crate::crd::{ /// /// Returns `(driver, executor)`: the operator-generated base arguments with the role's /// `jvmArgumentOverrides` applied on top. +/// +/// `product_version` is the resolved Spark product version (e.g. `4.1.2`); it gates the +/// version-specific OpenLineage `--add-opens` flag. pub fn construct_extra_java_options( spark_application: &SparkApplication, s3_conn: &Option, log_dir: &Option, -) -> (String, String) { + product_version: &str, +) -> Result<(String, String), Error> { // Note (@sbernauer): As of 2025-03-04, we did not set any heap related JVM arguments, so I // kept the implementation as is. We can always re-visit this as needed. @@ -36,6 +42,16 @@ 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. + // + // This is scoped to Spark 4.x: the flag is a no-op — and on some JVMs a startup error — on the + // JDK 11/17 Spark 3.x images the operator also ships, so it must not be emitted there. + if spark_application.openlineage_enabled() && spark_major_version(product_version)? >= 4 { + 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 @@ -56,16 +72,24 @@ pub fn construct_extra_java_options( None => jvm_args.clone(), }; - (driver.join(" "), executor.join(" ")) + Ok((driver.join(" "), executor.join(" "))) } #[cfg(test)] mod tests { + use rstest::rstest; + use super::*; + fn spark_app_from_yaml(input: &str) -> SparkApplication { + let deserializer = serde_yaml::Deserializer::from_str(input); + serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap() + } + #[test] fn test_construct_jvm_arguments_defaults() { - let input = r#" + let spark_app = spark_app_from_yaml( + r#" apiVersion: spark.stackable.tech/v1alpha1 kind: SparkApplication metadata: @@ -74,14 +98,11 @@ mod tests { mode: cluster mainApplicationFile: test.py sparkImage: - productVersion: 1.2.3 - "#; - - let deserializer = serde_yaml::Deserializer::from_str(input); - let spark_app: SparkApplication = - serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap(); + productVersion: 4.1.2 + "#, + ); let (driver_extra_java_options, executor_extra_java_options) = - construct_extra_java_options(&spark_app, &None, &None); + construct_extra_java_options(&spark_app, &None, &None, "4.1.2").unwrap(); assert_eq!( driver_extra_java_options, @@ -95,7 +116,8 @@ mod tests { #[test] fn test_construct_jvm_argument_overrides() { - let input = r#" + let spark_app = spark_app_from_yaml( + r#" apiVersion: spark.stackable.tech/v1alpha1 kind: SparkApplication metadata: @@ -104,7 +126,7 @@ mod tests { mode: cluster mainApplicationFile: test.py sparkImage: - productVersion: 1.2.3 + productVersion: 4.1.2 driver: jvmArgumentOverrides: add: @@ -115,13 +137,10 @@ mod tests { - -Dhttps.proxyHost=from-executor removeRegex: - -Djava.security.properties=.* - "#; - - 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); + construct_extra_java_options(&spark_app, &None, &None, "4.1.2").unwrap(); assert_eq!( driver_extra_java_options, @@ -132,4 +151,72 @@ mod tests { "-Dhttps.proxyHost=from-executor" ); } + + const OPENLINEAGE_ENABLED: &str = r#" + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-example + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: PLACEHOLDER + openLineage: + enabled: true + "#; + + const OPENLINEAGE_DISABLED: &str = r#" + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-example + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: PLACEHOLDER + openLineage: + enabled: false + "#; + + const OPENLINEAGE_ABSENT: &str = r#" + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-example + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: PLACEHOLDER + "#; + + /// `--add-opens` is emitted only on Spark 4.x with OpenLineage enabled — never on the Scala + /// 2.12 / JDK 17 Spark 3.5.x images, and never when OpenLineage is disabled or absent. + #[rstest] + #[case::enabled_spark_3(OPENLINEAGE_ENABLED, "3.5.8", false)] + #[case::enabled_spark_4(OPENLINEAGE_ENABLED, "4.1.2", true)] + #[case::disabled_spark_4(OPENLINEAGE_DISABLED, "4.1.2", false)] + #[case::absent_spark_4(OPENLINEAGE_ABSENT, "4.1.2", false)] + fn test_openlineage_add_opens_is_version_gated( + #[case] yaml_template: &str, + #[case] product_version: &str, + #[case] expect_add_opens: bool, + ) { + let spark_app = spark_app_from_yaml(&yaml_template.replace("PLACEHOLDER", product_version)); + let (driver_extra_java_options, executor_extra_java_options) = + construct_extra_java_options(&spark_app, &None, &None, product_version).unwrap(); + + assert_eq!( + driver_extra_java_options.contains(OPENLINEAGE_ADD_OPENS), + expect_add_opens, + "driver --add-opens presence mismatch for Spark {product_version}" + ); + assert_eq!( + executor_extra_java_options.contains(OPENLINEAGE_ADD_OPENS), + expect_add_opens, + "executor --add-opens presence mismatch for Spark {product_version}" + ); + } } diff --git a/rust/operator-binary/src/crd/constants.rs b/rust/operator-binary/src/crd/constants.rs index efef0f14..855fe87d 100644 --- a/rust/operator-binary/src/crd/constants.rs +++ b/rust/operator-binary/src/crd/constants.rs @@ -103,6 +103,32 @@ 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. +/// +/// This is a stable, version- and Scala-independent symlink the image maintains (mirroring the +/// `jmx_prometheus_javaagent.jar` pattern): the image bakes the correct `openlineage-spark_-.jar` +/// for its Spark build and symlinks it here. Referencing the symlink decouples the operator from the +/// jar's version and Scala suffix, so bumping either only touches `docker-images/spark-k8s`. +pub const OPENLINEAGE_JAR_LOCAL_URI: &str = + "local:///stackable/spark/openlineage/openlineage-spark.jar"; + +/// 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 +/// 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 { [ diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index a456b1d9..b9da7e25 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -9,7 +9,7 @@ use constants::*; use history::LogFileDirectorySpec; use logdir::ResolvedLogDir; use serde::{Deserialize, Serialize}; -use snafu::{OptionExt, ResultExt, Snafu}; +use snafu::{OptionExt, ResultExt, Snafu, ensure}; use stackable_operator::{ builder::pod::volume::{ SecretFormat, SecretOperatorVolumeSourceBuilder, SecretOperatorVolumeSourceBuilderError, @@ -114,6 +114,21 @@ pub enum Error { #[snafu(display("failed to configure log directory"))] ConfigureLogDir { source: logdir::Error }, + + #[snafu(display( + "OpenLineage is enabled but no backend endpoint could be resolved: set \ + `spec.openLineage.configMapName` (a discovery ConfigMap holding the `ADDRESS` key) or \ + `spark.openlineage.transport.url` in `sparkConf`" + ))] + MissingOpenLineageEndpoint, + + #[snafu(display( + "failed to parse the Spark major version from the resolved product version {product_version:?}" + ))] + UnparseableSparkVersion { + source: std::num::ParseIntError, + product_version: String, + }, } pub type SparkApplicationJobRoleType = @@ -248,6 +263,43 @@ pub mod versioned { /// The log file directory definition used by the Spark history server. #[serde(default, skip_serializing_if = "Option::is_none")] pub log_file_directory: Option, + + /// 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](DOCS_BASE_URL_PLACEHOLDER/spark-k8s/usage-guide/openlineage). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub open_lineage: Option, + } + + /// OpenLineage lineage emission for a [`SparkApplication`]. + /// + /// When enabled, the operator injects the OpenLineage Spark listener, points its HTTP transport + /// at the backend resolved from the discovery ConfigMap, and sets a stable job name. All injected + /// values are defaults: they can be overridden via `sparkConf`. + #[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + pub struct OpenLineageSpec { + /// Enable OpenLineage event emission. Defaults to `false` (nothing is injected). + #[serde(default)] + pub enabled: bool, + + /// The OpenLineage namespace lineage is reported under. + /// Defaults to the application's Kubernetes namespace (`metadata.namespace`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub namespace: Option, + + /// 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). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub app_name: Option, + + /// Name of the OpenLineage backend [discovery ConfigMap](DOCS_BASE_URL_PLACEHOLDER/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. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub config_map_name: Option, } #[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize, Merge)] @@ -260,6 +312,50 @@ pub mod versioned { } } +/// The resolved OpenLineage job/app name and where it came from. +/// See [`v1alpha1::SparkApplication::resolved_openlineage_app_name`]. +#[derive(Clone, Debug, PartialEq)] +pub struct ResolvedOpenLineageAppName { + /// The resolved, non-blank name written to `spark.openlineage.appName`. + pub name: String, + /// `true` when the name fell back to `metadata.name`; the controller then emits a warning event + /// because a per-run-unique name (e.g. an orchestrator-generated `-` suffix) fragments + /// backend run history. + pub from_metadata_name: bool, +} + +/// Appends `value` to a comma-separated `--conf` value in `submit_conf`, preserving any existing +/// (e.g. user-supplied) entries and skipping `value` if it is already present. Used for the +/// OpenLineage keys that must accumulate rather than clobber (`spark.extraListeners`, `spark.jars`). +fn append_conf_csv(submit_conf: &mut BTreeMap, key: &str, value: &str) { + match submit_conf.get_mut(key) { + Some(existing) if !existing.is_empty() => { + if !existing.split(',').any(|entry| entry.trim() == value) { + existing.push(','); + existing.push_str(value); + } + } + _ => { + submit_conf.insert(key.to_string(), value.to_string()); + } + } +} + +/// Parses the Spark major version from a resolved product version string such as `4.1.2` or +/// `3.5.8`. Used to gate version-specific OpenLineage wiring: the `--add-opens` JVM flag (only +/// needed on the JDK 17+/Spark 4.x line) and the Scala build of the baked OpenLineage jar +/// (Stackable ships Spark 4.x as Scala 2.13 and Spark 3.5.x as Scala 2.12). +pub(crate) fn spark_major_version(product_version: &str) -> Result { + product_version + .split('.') + .next() + .unwrap_or_default() + .parse::() + .context(UnparseableSparkVersionSnafu { + product_version: product_version.to_string(), + }) +} + impl v1alpha1::SparkApplication { /// Returns if this [`SparkApplication`] has already created a Kubernetes Job doing the actual `spark-submit`. /// @@ -556,11 +652,22 @@ impl v1alpha1::SparkApplication { mounts } + /// True when OpenLineage emission is configured and enabled. Off is expressible two ways: + /// the `openLineage` block absent, or present with `enabled: false`. + pub fn openlineage_enabled(&self) -> bool { + self.spec + .open_lineage + .as_ref() + .is_some_and(|open_lineage| open_lineage.enabled) + } + pub fn build_command( &self, s3conn: &Option, log_dir: &Option, spark_image: &str, + product_version: &str, + openlineage_endpoint: Option<&str>, ) -> Result, Error> { // mandatory properties let mode = &self.spec.mode; @@ -659,7 +766,7 @@ impl v1alpha1::SparkApplication { } let (driver_extra_java_options, executor_extra_java_options) = - construct_extra_java_options(self, s3conn, log_dir); + construct_extra_java_options(self, s3conn, log_dir, product_version)?; submit_cmd.extend(vec![ format!("--conf spark.driver.extraJavaOptions=\"{driver_extra_java_options}\""), format!("--conf spark.executor.extraJavaOptions=\"{executor_extra_java_options}\""), @@ -733,9 +840,63 @@ impl v1alpha1::SparkApplication { submit_cmd.push(format!("--conf spark.jars.ivy={VOLUME_MOUNT_PATH_IVY2}")) } + // OpenLineage: inject the transport + job-name config that can be overridden by the user's + // `sparkConf` (so it goes in BEFORE the merge below). The two append-merge keys + // (`spark.jars`, `spark.extraListeners`) are handled AFTER the merge so a user value is + // combined with — not clobbered by — ours. See the OpenLineage usage guide. + if let Some(open_lineage) = &self.spec.open_lineage + && open_lineage.enabled + { + // Fail fast rather than emit a transport that silently drops every event. + let has_url_override = self + .spec + .spark_conf + .contains_key("spark.openlineage.transport.url"); + ensure!( + openlineage_endpoint.is_some() || has_url_override, + MissingOpenLineageEndpointSnafu + ); + + submit_conf.insert( + "spark.openlineage.transport.type".to_string(), + "http".to_string(), + ); + if let Some(endpoint) = openlineage_endpoint { + submit_conf.insert( + "spark.openlineage.transport.url".to_string(), + endpoint.to_string(), + ); + } + + let namespace = match &open_lineage.namespace { + Some(namespace) => namespace.clone(), + None => self.metadata.namespace.clone().context(NoNamespaceSnafu)?, + }; + submit_conf.insert("spark.openlineage.namespace".to_string(), namespace); + + // Stable job name — fixes the intermittent `unknown` bug (see the usage guide). + submit_conf.insert( + "spark.openlineage.appName".to_string(), + self.resolved_openlineage_app_name()?.name, + ); + } + // conf arguments: these should follow - and thus override - values set from resource limits above submit_conf.extend(self.spec.spark_conf.clone()); + // OpenLineage append-merge keys: combine our values with any user-provided ones (already + // merged into `submit_conf` above) so neither clobbers the other. + if self.openlineage_enabled() { + append_conf_csv( + &mut submit_conf, + "spark.extraListeners", + OPENLINEAGE_LISTENER_CLASS, + ); + // Reference the stable symlink the image maintains; it points at the correct + // Scala/version build for this image, so the operator needs to know neither. + append_conf_csv(&mut submit_conf, "spark.jars", OPENLINEAGE_JAR_LOCAL_URI); + } + // ...before being added to the command collection for (key, value) in submit_conf { submit_cmd.push(format!("--conf \"{key}={value}\"")); @@ -756,6 +917,39 @@ impl v1alpha1::SparkApplication { Ok(vec![submit_cmd.join(" ")]) } + /// Resolves the stable OpenLineage job/app name and its provenance (MVP §5), in priority order: + /// 1. `spec.openLineage.appName`, else + /// 2. `spark.app.name` from `sparkConf`, else + /// 3. `metadata.name` (a fallback the controller flags with a warning event, because a + /// per-run-unique name would fragment backend run history). + /// + /// Always yields a non-blank name — which is exactly what fixes the intermittent `unknown` bug. + pub fn resolved_openlineage_app_name(&self) -> Result { + if let Some(app_name) = self + .spec + .open_lineage + .as_ref() + .and_then(|open_lineage| open_lineage.app_name.clone()) + { + return Ok(ResolvedOpenLineageAppName { + name: app_name, + from_metadata_name: false, + }); + } + + if let Some(app_name) = self.spec.spark_conf.get("spark.app.name") { + return Ok(ResolvedOpenLineageAppName { + name: app_name.clone(), + from_metadata_name: false, + }); + } + + Ok(ResolvedOpenLineageAppName { + name: self.metadata.name.clone().context(ObjectHasNoNameSnafu)?, + from_metadata_name: true, + }) + } + pub fn env( &self, s3conn: &Option, @@ -1574,6 +1768,301 @@ spec: assert_eq!(got, expected); } + /// Builds a `SparkApplication` from YAML and returns the assembled spark-submit command string. + /// `product_version` stands in for the resolved Spark product version (gates the version-specific + /// OpenLineage wiring) and `openlineage_endpoint` for the value the controller resolves from the + /// discovery ConfigMap. + fn build_command_with_openlineage( + yaml: &str, + product_version: &str, + openlineage_endpoint: Option<&str>, + ) -> String { + let spark_application = serde_yaml::from_str::(yaml).unwrap(); + spark_application + .build_command( + &None, + &None, + "test-image", + product_version, + openlineage_endpoint, + ) + .unwrap() + .join(" ") + } + + const OPENLINEAGE_ENABLED_APP: &str = indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + "#}; + + #[test] + fn test_openlineage_injects_conf_when_enabled() { + let command = build_command_with_openlineage( + OPENLINEAGE_ENABLED_APP, + "4.1.2", + Some("http://marquez:5000"), + ); + + assert!(command.contains(r#"--conf "spark.openlineage.transport.type=http""#)); + assert!( + command.contains(r#"--conf "spark.openlineage.transport.url=http://marquez:5000""#) + ); + // Namespace defaults to metadata.namespace. + assert!(command.contains(r#"--conf "spark.openlineage.namespace=default""#)); + // appName falls back to metadata.name (no appName / spark.app.name set). + assert!(command.contains(r#"--conf "spark.openlineage.appName=spark-examples""#)); + assert!(command.contains(&format!( + r#"--conf "spark.extraListeners={OPENLINEAGE_LISTENER_CLASS}""# + ))); + // The jar is referenced via the stable, Scala/version-independent image symlink. + assert!(command.contains(&format!( + r#"--conf "spark.jars={OPENLINEAGE_JAR_LOCAL_URI}""# + ))); + // --add-opens reaches both driver and executor on Spark 4.x. + assert_eq!( + command.matches(OPENLINEAGE_ADD_OPENS).count(), + 2, + "expected --add-opens on both driver and executor extraJavaOptions" + ); + } + + #[test] + fn test_openlineage_stable_jar_uri_and_no_add_opens_on_spark_3() { + // On the JDK 17 Spark 3.5.x images the operator references the same stable jar symlink (the + // image points it at the Scala 2.12 build) and must NOT emit `--add-opens`. + let command = build_command_with_openlineage( + OPENLINEAGE_ENABLED_APP, + "3.5.8", + Some("http://marquez:5000"), + ); + + assert!(command.contains(&format!( + r#"--conf "spark.jars={OPENLINEAGE_JAR_LOCAL_URI}""# + ))); + assert!( + !command.contains(OPENLINEAGE_ADD_OPENS), + "--add-opens must not be emitted on Spark 3.x" + ); + // The rest of the OpenLineage wiring is still present on Spark 3.x. + assert!(command.contains(&format!( + r#"--conf "spark.extraListeners={OPENLINEAGE_LISTENER_CLASS}""# + ))); + assert!(command.contains(r#"--conf "spark.openlineage.transport.type=http""#)); + } + + #[rstest] + #[case::absent(indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + "#})] + #[case::disabled(indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: false + "#})] + fn test_openlineage_injects_nothing_when_absent_or_disabled(#[case] yaml: &str) { + // Endpoint is supplied to prove it is ignored, not that it is simply missing. + let command = build_command_with_openlineage(yaml, "4.1.2", Some("http://marquez:5000")); + + assert!(!command.contains("openlineage")); + assert!(!command.contains("OpenLineageSparkListener")); + assert!(!command.contains("--add-opens")); + assert!(!command.contains("spark.jars=")); + } + + #[test] + fn test_openlineage_appends_to_existing_extra_listeners() { + let yaml = indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + sparkConf: + spark.extraListeners: com.example.CustomListener + "#}; + let command = build_command_with_openlineage(yaml, "4.1.2", Some("http://marquez:5000")); + + assert!(command.contains(&format!( + r#"--conf "spark.extraListeners=com.example.CustomListener,{OPENLINEAGE_LISTENER_CLASS}""# + ))); + } + + #[test] + fn test_openlineage_conf_is_overridable_via_spark_conf() { + let yaml = indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + sparkConf: + spark.openlineage.transport.url: http://custom:1234 + spark.openlineage.namespace: custom-ns + "#}; + let command = build_command_with_openlineage(yaml, "4.1.2", Some("http://marquez:5000")); + + assert!(command.contains(r#"--conf "spark.openlineage.transport.url=http://custom:1234""#)); + assert!(!command.contains("http://marquez:5000")); + assert!(command.contains(r#"--conf "spark.openlineage.namespace=custom-ns""#)); + } + + #[test] + fn test_openlineage_fails_fast_without_endpoint() { + // Enabled, but neither a resolved endpoint nor a sparkConf transport.url override. + let spark_application = + serde_yaml::from_str::(OPENLINEAGE_ENABLED_APP).unwrap(); + let result = spark_application.build_command(&None, &None, "test-image", "4.1.2", None); + + assert!(matches!(result, Err(Error::MissingOpenLineageEndpoint))); + } + + #[test] + fn test_openlineage_endpoint_from_spark_conf_satisfies_fail_fast() { + // No resolved endpoint, but a sparkConf override → must NOT fail. + let yaml = indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + sparkConf: + spark.openlineage.transport.url: http://custom:1234 + "#}; + let command = build_command_with_openlineage(yaml, "4.1.2", None); + + assert!(command.contains(r#"--conf "spark.openlineage.transport.url=http://custom:1234""#)); + } + + #[rstest] + #[case::explicit_app_name( + indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: metadata-name + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + appName: explicit-name + sparkConf: + spark.app.name: spark-conf-name + "#}, + "explicit-name", + false + )] + #[case::spark_app_name( + indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: metadata-name + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + sparkConf: + spark.app.name: spark-conf-name + "#}, + "spark-conf-name", + false + )] + #[case::metadata_name_fallback( + indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: metadata-name + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + "#}, + "metadata-name", + true + )] + fn test_openlineage_app_name_resolution( + #[case] yaml: &str, + #[case] expected_name: &str, + #[case] expected_from_metadata_name: bool, + ) { + let spark_application = serde_yaml::from_str::(yaml).unwrap(); + let resolved = spark_application.resolved_openlineage_app_name().unwrap(); + + assert_eq!(resolved.name, expected_name); + assert_eq!(resolved.from_metadata_name, expected_from_metadata_name); + } + impl RoundtripTestData for v1alpha1::SparkApplicationSpec { fn roundtrip_test_data() -> Vec { stackable_operator::utils::yaml_from_str_singleton_map(indoc! {r#" diff --git a/rust/operator-binary/src/crd/template_merger.rs b/rust/operator-binary/src/crd/template_merger.rs index 7f8afbba..b12bff0f 100644 --- a/rust/operator-binary/src/crd/template_merger.rs +++ b/rust/operator-binary/src/crd/template_merger.rs @@ -55,6 +55,11 @@ pub fn deep_merge(base: &SparkApplication, overlay: &SparkApplication) -> SparkA .vector_aggregator_config_map_name .clone() .or_else(|| base.spec.vector_aggregator_config_map_name.clone()), + open_lineage: overlay + .spec + .open_lineage + .clone() + .or_else(|| base.spec.open_lineage.clone()), s3connection: overlay .spec .s3connection diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs index 9f14b776..c91b2290 100644 --- a/rust/operator-binary/src/main.rs +++ b/rust/operator-binary/src/main.rs @@ -70,6 +70,7 @@ struct Opts { pub struct Ctx { pub client: stackable_operator::client::Client, pub operator_environment: OperatorEnvironmentOptions, + pub event_recorder: Recorder, } #[tokio::main] @@ -136,11 +137,6 @@ async fn main() -> anyhow::Result<()> { .run(sigterm_watcher.handle()) .map_err(|err| anyhow!(err).context("failed to run webhook server")); - let ctx = Ctx { - client: client.clone(), - operator_environment: operator_environment.clone(), - }; - let spark_event_recorder = Arc::new(Recorder::new( client.as_kube_client(), Reporter { @@ -148,6 +144,14 @@ async fn main() -> anyhow::Result<()> { instance: None, }, )); + + let ctx = Ctx { + client: client.clone(), + operator_environment: operator_environment.clone(), + // Shares the recorder (and thus its event-aggregation cache) with the + // report_controller_reconciled reporting stream below; `Recorder` is `Clone`. + event_recorder: (*spark_event_recorder).clone(), + }; let app_controller = Controller::new( watch_namespace .get_api::>(&client), @@ -224,11 +228,6 @@ async fn main() -> anyhow::Result<()> { }, ).map(anyhow::Ok); - // Create new object because Ctx cannot be cloned - let ctx = Ctx { - client: client.clone(), - operator_environment: operator_environment.clone(), - }; let history_event_recorder = Arc::new(Recorder::new( client.as_kube_client(), Reporter { @@ -236,6 +235,12 @@ async fn main() -> anyhow::Result<()> { instance: None, }, )); + // Create new object because Ctx cannot be cloned + let ctx = Ctx { + client: client.clone(), + operator_environment: operator_environment.clone(), + event_recorder: (*history_event_recorder).clone(), + }; let history_controller = Controller::new( watch_namespace .get_api::>( @@ -289,11 +294,6 @@ async fn main() -> anyhow::Result<()> { .map(anyhow::Ok); // ============================== - // Create new object because Ctx cannot be cloned - let ctx = Ctx { - client: client.clone(), - operator_environment, - }; let connect_event_recorder = Arc::new(Recorder::new( client.as_kube_client(), Reporter { @@ -301,6 +301,12 @@ async fn main() -> anyhow::Result<()> { instance: None, }, )); + // Create new object because Ctx cannot be cloned + let ctx = Ctx { + client: client.clone(), + operator_environment, + event_recorder: (*connect_event_recorder).clone(), + }; let connect_controller = Controller::new( watch_namespace .get_api::>( diff --git a/rust/operator-binary/src/spark_k8s_controller.rs b/rust/operator-binary/src/spark_k8s_controller.rs index a7a128b7..c0f34c3f 100644 --- a/rust/operator-binary/src/spark_k8s_controller.rs +++ b/rust/operator-binary/src/spark_k8s_controller.rs @@ -1,12 +1,16 @@ use std::sync::Arc; -use snafu::{ResultExt, Snafu}; +use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ builder::{self}, + k8s_openapi::api::core::v1::ConfigMap, kube::{ - ResourceExt, + Resource, ResourceExt, core::{DeserializeGuard, error_boundary}, - runtime::controller::Action, + runtime::{ + controller::Action, + events::{Event, EventType, Recorder}, + }, }, logging::controller::ReconcilerError, shared::time::Duration, @@ -68,6 +72,18 @@ pub enum Error { #[snafu(display("vector agent is enabled but vector aggregator ConfigMap is missing"))] VectorAggregatorConfigMapMissing, + #[snafu(display("failed to resolve the OpenLineage discovery ConfigMap {name:?}"))] + ResolveOpenLineageConfigMap { + source: stackable_operator::client::Error, + name: String, + }, + + #[snafu(display( + "the OpenLineage discovery ConfigMap {name:?} does not contain the required key \ + {OPENLINEAGE_CONFIG_MAP_ADDRESS_KEY:?}" + ))] + OpenLineageConfigMapMissingAddress { name: String }, + #[snafu(display("failed to validate the logging configuration"))] ValidateLoggingConfig { source: stackable_operator::v2::product_logging::framework::Error, @@ -223,8 +239,31 @@ pub async fn reconcile( .await .context(ApplyApplicationSnafu)?; + // Resolve the OpenLineage backend endpoint from the discovery ConfigMap (if configured) and + // warn if the job name falls back to `metadata.name` — both only matter when OpenLineage is + // enabled. + let openlineage_endpoint = resolve_openlineage_endpoint(client, spark_application).await?; + if spark_application + .spec + .open_lineage + .as_ref() + .is_some_and(|open_lineage| open_lineage.enabled) + && spark_application + .resolved_openlineage_app_name() + .context(BuildCommandSnafu)? + .from_metadata_name + { + publish_openlineage_app_name_warning(&ctx.event_recorder, spark_application).await; + } + let job_commands = spark_application - .build_command(opt_s3conn, logdir, &resolved_product_image.image) + .build_command( + opt_s3conn, + logdir, + &resolved_product_image.image, + &resolved_product_image.product_version, + openlineage_endpoint.as_deref(), + ) .context(BuildCommandSnafu)?; let submit_config = spark_application @@ -291,3 +330,75 @@ pub fn error_policy( _ => Action::requeue(*Duration::from_secs(5)), } } + +/// Resolves the OpenLineage backend base URL from the discovery ConfigMap referenced by +/// `spec.openLineage.configMapName` (key [`OPENLINEAGE_CONFIG_MAP_ADDRESS_KEY`]). Returns `None` +/// when OpenLineage is disabled/absent or no ConfigMap is referenced — in the latter case a +/// `sparkConf` override is expected, which `build_command` enforces. +async fn resolve_openlineage_endpoint( + client: &stackable_operator::client::Client, + spark_application: &v1alpha1::SparkApplication, +) -> Result> { + let Some(open_lineage) = &spark_application.spec.open_lineage else { + return Ok(None); + }; + let (true, Some(config_map_name)) = (open_lineage.enabled, &open_lineage.config_map_name) + else { + return Ok(None); + }; + + let name = config_map_name.as_ref(); + let namespace = spark_application.namespace().unwrap_or_default(); + + let config_map = client.get::(name, &namespace).await.context( + ResolveOpenLineageConfigMapSnafu { + name: name.to_string(), + }, + )?; + + let address = config_map + .data + .as_ref() + .and_then(|data| data.get(OPENLINEAGE_CONFIG_MAP_ADDRESS_KEY)) + .cloned() + .context(OpenLineageConfigMapMissingAddressSnafu { + name: name.to_string(), + })?; + + Ok(Some(address)) +} + +/// Emits a Kubernetes warning event that the OpenLineage job name fell back to `metadata.name`, +/// which fragments backend run history when that name is unique per run. Event-publishing failures +/// are logged, not propagated — they must not fail reconciliation. +async fn publish_openlineage_app_name_warning( + event_recorder: &Recorder, + spark_application: &v1alpha1::SparkApplication, +) { + let name = spark_application.name_any(); + let publish_result = event_recorder + .publish( + &Event { + type_: EventType::Warning, + reason: "OpenLineageAppNameFallback".into(), + note: Some(format!( + "OpenLineage job name falls back to metadata.name ({name:?}) because neither \ + spec.openLineage.appName nor spark.app.name is set. If metadata.name is unique \ + per run (e.g. an orchestrator-generated - suffix), backend run \ + history will be fragmented into a new job per run. Set \ + spec.openLineage.appName to a stable value to avoid this." + )), + action: "ResolveOpenLineageAppName".into(), + secondary: None, + }, + &spark_application.object_ref(&()), + ) + .await; + + if let Err(error) = publish_result { + tracing::warn!( + ?error, + "failed to publish OpenLineage app-name fallback warning event" + ); + } +} diff --git a/tests/templates/kuttl/hbase-connector/10-deploy-spark-app.yaml.j2 b/tests/templates/kuttl/hbase-connector/10-deploy-spark-app.yaml.j2 index 76245c96..87aa996d 100644 --- a/tests/templates/kuttl/hbase-connector/10-deploy-spark-app.yaml.j2 +++ b/tests/templates/kuttl/hbase-connector/10-deploy-spark-app.yaml.j2 @@ -77,7 +77,11 @@ data: spark = SparkSession.builder.appName("test-hbase").getOrCreate() df = spark.createDataFrame( - [("row1", "Hello, Stackable!")], + [ + ("row1", "Hello, Stackable!"), + ("row2", "Hello, HBase!"), + ("row3", "Hello, Spark!"), + ], "key: string, value: string" ) @@ -101,3 +105,53 @@ data: .option('catalog', catalog)\ .option('newtable', '5')\ .save() + + # Read the whole table back and register it as a temporary view so we can + # run SQL queries against it. + # + # hbase.spark.pushdown.columnfilter=false disables server-side predicate + # pushdown: the connector would otherwise ship a SparkSQLPushDownFilter to + # the RegionServers, which requires the hbase-spark connector classes on + # the RegionServer classpath (not present in the Stackable HBase image). + # With pushdown off, WHERE clauses are evaluated Spark-side over a full + # scan, which is all this test needs. + read_df = spark\ + .read\ + .format("org.apache.hadoop.hbase.spark")\ + .option('catalog', catalog)\ + .option('hbase.spark.pushdown.columnfilter', 'false')\ + .load() + read_df.createOrReplaceTempView("test_hbase") + + # Full scan: all rows that were written must be readable again. + all_rows = spark.sql("SELECT key, value FROM test_hbase").collect() + actual = {row["key"]: row["value"] for row in all_rows} + expected = { + "row1": "Hello, Stackable!", + "row2": "Hello, HBase!", + "row3": "Hello, Spark!", + } + assert actual == expected, f"full scan mismatch: {actual} != {expected}" + + # Row-key point lookup. + point = spark.sql("SELECT value FROM test_hbase WHERE key = 'row2'").collect() + assert len(point) == 1, f"point lookup returned {len(point)} rows, expected 1" + assert point[0]["value"] == "Hello, HBase!", f"unexpected value: {point[0]['value']}" + + # Range scan on the row key. + range_rows = spark.sql( + "SELECT key FROM test_hbase WHERE key >= 'row2' ORDER BY key" + ).collect() + assert [r["key"] for r in range_rows] == ["row2", "row3"], \ + f"unexpected range scan result: {[r['key'] for r in range_rows]}" + + # Filter on a non-rowkey column value. + like_rows = spark.sql( + "SELECT key FROM test_hbase WHERE value LIKE 'Hello, S%' ORDER BY key" + ).collect() + assert [r["key"] for r in like_rows] == ["row1", "row3"], \ + f"unexpected value filter result: {[r['key'] for r in like_rows]}" + + # Aggregation over the scanned rows. + count = spark.sql("SELECT COUNT(*) AS c FROM test_hbase").collect()[0]["c"] + assert count == 3, f"expected 3 rows, got {count}" diff --git a/tests/templates/kuttl/product-config-compat/10-deploy-spark-app.yaml.j2 b/tests/templates/kuttl/product-config-compat/10-deploy-spark-app.yaml.j2 index b2b9104d..9c4d46a7 100644 --- a/tests/templates/kuttl/product-config-compat/10-deploy-spark-app.yaml.j2 +++ b/tests/templates/kuttl/product-config-compat/10-deploy-spark-app.yaml.j2 @@ -5,6 +5,7 @@ metadata: name: pyspark-pi spec: sparkImage: + pullPolicy: IfNotPresent {% if test_scenario['values']['spark-regression'].find(",") > 0 %} custom: "{{ test_scenario['values']['spark-regression'].split(',')[1] }}" productVersion: "{{ test_scenario['values']['spark-regression'].split(',')[0] }}" diff --git a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-driver-pod-template-data.json b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-driver-pod-template-data.json index c65e0c89..1e35fafb 100644 --- a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-driver-pod-template-data.json +++ b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-driver-pod-template-data.json @@ -2,5 +2,5 @@ "log4j2.properties": "appenders = FILE, CONSOLE\n\nappender.CONSOLE.type = Console\nappender.CONSOLE.name = CONSOLE\nappender.CONSOLE.target = SYSTEM_ERR\nappender.CONSOLE.layout.type = PatternLayout\nappender.CONSOLE.layout.pattern = %d{ISO8601} %p [%t] %c - %m%n\nappender.CONSOLE.filter.threshold.type = ThresholdFilter\nappender.CONSOLE.filter.threshold.level = INFO\n\nappender.FILE.type = RollingFile\nappender.FILE.name = FILE\nappender.FILE.fileName = /stackable/log/spark/spark.log4j2.xml\nappender.FILE.filePattern = /stackable/log/spark/spark.log4j2.xml.%i\nappender.FILE.layout.type = XMLLayout\nappender.FILE.policies.type = Policies\nappender.FILE.policies.size.type = SizeBasedTriggeringPolicy\nappender.FILE.policies.size.size = 5MB\nappender.FILE.strategy.type = DefaultRolloverStrategy\nappender.FILE.strategy.max = 1\nappender.FILE.filter.threshold.type = ThresholdFilter\nappender.FILE.filter.threshold.level = INFO\n\n\nrootLogger.level=INFO\nrootLogger.appenderRefs = CONSOLE, FILE\nrootLogger.appenderRef.CONSOLE.ref = CONSOLE\nrootLogger.appenderRef.FILE.ref = FILE", "security.properties": "networkaddress.cache.negative.ttl=0\nnetworkaddress.cache.ttl=30\n", "spark-env.sh": "", - "template.yaml": "metadata:\n labels:\n app.kubernetes.io/component: spark\n app.kubernetes.io/instance: pyspark-pi\n app.kubernetes.io/managed-by: spark.stackable.tech_sparkapplication\n app.kubernetes.io/name: spark-k8s\n app.kubernetes.io/role-group: sparkapplication\n app.kubernetes.io/version: 3.5.8-stackable0.0.0-dev\n prometheus.io/scrape: 'true'\n stackable.tech/vendor: Stackable\n name: spark\nspec:\n affinity: {}\n containers:\n - env:\n - name: CONTAINERDEBUG_LOG_DIRECTORY\n value: /stackable/log/containerdebug\n - name: _STACKABLE_PRE_HOOK\n value: containerdebug --output=/stackable/log/containerdebug-state.json --loop &\n image: oci.stackable.tech/sdp/spark-k8s:3.5.8-stackable0.0.0-dev\n imagePullPolicy: Always\n name: spark\n resources:\n limits:\n cpu: '2'\n memory: 1Gi\n requests:\n cpu: '1'\n memory: 1Gi\n volumeMounts:\n - mountPath: /stackable/log_config\n name: log-config\n - mountPath: /stackable/log\n name: log\n enableServiceLinks: false\n securityContext:\n fsGroup: 1000\n serviceAccountName: pyspark-pi\n volumes:\n - emptyDir:\n sizeLimit: 39Mi\n name: log\n - configMap:\n name: pyspark-pi-driver-pod-template\n name: log-config\n - configMap:\n name: pyspark-pi-driver-pod-template\n name: config\n" + "template.yaml": "metadata:\n labels:\n app.kubernetes.io/component: spark\n app.kubernetes.io/instance: pyspark-pi\n app.kubernetes.io/managed-by: spark.stackable.tech_sparkapplication\n app.kubernetes.io/name: spark-k8s\n app.kubernetes.io/role-group: sparkapplication\n app.kubernetes.io/version: 3.5.8-stackable0.0.0-dev\n prometheus.io/scrape: 'true'\n stackable.tech/vendor: Stackable\n name: spark\nspec:\n affinity: {}\n containers:\n - env:\n - name: CONTAINERDEBUG_LOG_DIRECTORY\n value: /stackable/log/containerdebug\n - name: _STACKABLE_PRE_HOOK\n value: containerdebug --output=/stackable/log/containerdebug-state.json --loop &\n image: oci.stackable.tech/sdp/spark-k8s:3.5.8-stackable0.0.0-dev\n imagePullPolicy: IfNotPresent\n name: spark\n resources:\n limits:\n cpu: '2'\n memory: 1Gi\n requests:\n cpu: '1'\n memory: 1Gi\n volumeMounts:\n - mountPath: /stackable/log_config\n name: log-config\n - mountPath: /stackable/log\n name: log\n enableServiceLinks: false\n securityContext:\n fsGroup: 1000\n serviceAccountName: pyspark-pi\n volumes:\n - emptyDir:\n sizeLimit: 39Mi\n name: log\n - configMap:\n name: pyspark-pi-driver-pod-template\n name: log-config\n - configMap:\n name: pyspark-pi-driver-pod-template\n name: config\n" } diff --git a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-executor-pod-template-data.json b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-executor-pod-template-data.json index 03c53b63..096896c5 100644 --- a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-executor-pod-template-data.json +++ b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-executor-pod-template-data.json @@ -2,5 +2,5 @@ "log4j2.properties": "appenders = FILE, CONSOLE\n\nappender.CONSOLE.type = Console\nappender.CONSOLE.name = CONSOLE\nappender.CONSOLE.target = SYSTEM_ERR\nappender.CONSOLE.layout.type = PatternLayout\nappender.CONSOLE.layout.pattern = %d{ISO8601} %p [%t] %c - %m%n\nappender.CONSOLE.filter.threshold.type = ThresholdFilter\nappender.CONSOLE.filter.threshold.level = INFO\n\nappender.FILE.type = RollingFile\nappender.FILE.name = FILE\nappender.FILE.fileName = /stackable/log/spark/spark.log4j2.xml\nappender.FILE.filePattern = /stackable/log/spark/spark.log4j2.xml.%i\nappender.FILE.layout.type = XMLLayout\nappender.FILE.policies.type = Policies\nappender.FILE.policies.size.type = SizeBasedTriggeringPolicy\nappender.FILE.policies.size.size = 5MB\nappender.FILE.strategy.type = DefaultRolloverStrategy\nappender.FILE.strategy.max = 1\nappender.FILE.filter.threshold.type = ThresholdFilter\nappender.FILE.filter.threshold.level = INFO\n\n\nrootLogger.level=INFO\nrootLogger.appenderRefs = CONSOLE, FILE\nrootLogger.appenderRef.CONSOLE.ref = CONSOLE\nrootLogger.appenderRef.FILE.ref = FILE", "security.properties": "networkaddress.cache.negative.ttl=0\nnetworkaddress.cache.ttl=30\n", "spark-env.sh": "", - "template.yaml": "metadata:\n labels:\n app.kubernetes.io/component: spark\n app.kubernetes.io/instance: pyspark-pi\n app.kubernetes.io/managed-by: spark.stackable.tech_sparkapplication\n app.kubernetes.io/name: spark-k8s\n app.kubernetes.io/role-group: sparkapplication\n app.kubernetes.io/version: 3.5.8-stackable0.0.0-dev\n stackable.tech/vendor: Stackable\n name: spark\nspec:\n affinity: {}\n containers:\n - env:\n - name: CONTAINERDEBUG_LOG_DIRECTORY\n value: /stackable/log/containerdebug\n - name: _STACKABLE_PRE_HOOK\n value: containerdebug --output=/stackable/log/containerdebug-state.json --loop &\n image: oci.stackable.tech/sdp/spark-k8s:3.5.8-stackable0.0.0-dev\n imagePullPolicy: Always\n name: spark\n resources:\n limits:\n cpu: '2'\n memory: 1Gi\n requests:\n cpu: '1'\n memory: 1Gi\n volumeMounts:\n - mountPath: /stackable/log_config\n name: log-config\n - mountPath: /stackable/log\n name: log\n enableServiceLinks: false\n securityContext:\n fsGroup: 1000\n serviceAccountName: pyspark-pi\n volumes:\n - emptyDir:\n sizeLimit: 39Mi\n name: log\n - configMap:\n name: pyspark-pi-executor-pod-template\n name: log-config\n - configMap:\n name: pyspark-pi-executor-pod-template\n name: config\n" + "template.yaml": "metadata:\n labels:\n app.kubernetes.io/component: spark\n app.kubernetes.io/instance: pyspark-pi\n app.kubernetes.io/managed-by: spark.stackable.tech_sparkapplication\n app.kubernetes.io/name: spark-k8s\n app.kubernetes.io/role-group: sparkapplication\n app.kubernetes.io/version: 3.5.8-stackable0.0.0-dev\n stackable.tech/vendor: Stackable\n name: spark\nspec:\n affinity: {}\n containers:\n - env:\n - name: CONTAINERDEBUG_LOG_DIRECTORY\n value: /stackable/log/containerdebug\n - name: _STACKABLE_PRE_HOOK\n value: containerdebug --output=/stackable/log/containerdebug-state.json --loop &\n image: oci.stackable.tech/sdp/spark-k8s:3.5.8-stackable0.0.0-dev\n imagePullPolicy: IfNotPresent\n name: spark\n resources:\n limits:\n cpu: '2'\n memory: 1Gi\n requests:\n cpu: '1'\n memory: 1Gi\n volumeMounts:\n - mountPath: /stackable/log_config\n name: log-config\n - mountPath: /stackable/log\n name: log\n enableServiceLinks: false\n securityContext:\n fsGroup: 1000\n serviceAccountName: pyspark-pi\n volumes:\n - emptyDir:\n sizeLimit: 39Mi\n name: log\n - configMap:\n name: pyspark-pi-executor-pod-template\n name: log-config\n - configMap:\n name: pyspark-pi-executor-pod-template\n name: config\n" } diff --git a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-job-template-spec.json b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-job-template-spec.json index dc9ada45..22518c52 100644 --- a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-job-template-spec.json +++ b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-job-template-spec.json @@ -31,7 +31,7 @@ } ], "image": "oci.stackable.tech/sdp/spark-k8s:3.5.8-stackable0.0.0-dev", - "imagePullPolicy": "Always", + "imagePullPolicy": "IfNotPresent", "name": "spark-submit", "resources": { "limits": { diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml index 005ca2a1..5d387b63 100644 --- a/tests/test-definition.yaml +++ b/tests/test-definition.yaml @@ -56,7 +56,7 @@ dimensions: - "CLUSTER.LOCAL" - name: hbase values: - - 2.6.4 + - 2.6.6 - name: hdfs-latest values: - 3.5.0