diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 29e464030..fcd7f7746 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -24,8 +24,27 @@ on: jobs: test-agent-reader-and-model: - name: Test Agent, Reader and Model - runs-on: ubuntu-latest + name: ${{ matrix.name }} + runs-on: ubuntu-22.04 + strategy: + fail-fast: false + matrix: + include: + - name: Test Agent, Reader and Model (Spark 3, Java 8) + java-version: '8' + test-command: testAgentReaderModelSpark3 + doc-command: docAgentReaderModelSpark3 + + - name: Test Agent, Reader and Model (Spark 3, Java 11) + java-version: '11' + test-command: testAgentReaderModelSpark3 + doc-command: docAgentReaderModelSpark3 + + - name: Test Agent, Reader and Model (Spark 4, Java 17) + java-version: '17' + test-command: testAgentReaderModelSpark4 + doc-command: docAgentReaderModelSpark4 + steps: - name: Checkout code uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 @@ -34,20 +53,21 @@ jobs: persist-credentials: false - uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + - name: Setup JDK ${{ matrix.java-version }} + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.8" + distribution: temurin + java-version: ${{ matrix.java-version }} - name: Build and run tests - run: sbt testAllStandard + run: sbt ${{ matrix.test-command }} - - name: Generate documenation - run: sbt doc + - name: Generate documentation + run: sbt ${{ matrix.doc-command }} test-database-and-server: - name: Test Database and Server - runs-on: ubuntu-latest + name: Test Database and Server (Java 11) + runs-on: ubuntu-22.04 services: postgres: image: postgres:15 @@ -69,10 +89,11 @@ jobs: persist-credentials: false - uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + - name: Setup JDK 11 + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.11.0-11" + distribution: temurin + java-version: '11' - name: Build and run unit tests run: sbt "project database" test "project server" test @@ -88,7 +109,7 @@ jobs: sbom-and-aquasec-manifest: name: SBOM and Aquasec Manifest - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 steps: - name: Checkout code uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 diff --git a/.github/workflows/format_check.yml b/.github/workflows/format_check.yml index c60f7cb5c..43bcafa69 100644 --- a/.github/workflows/format_check.yml +++ b/.github/workflows/format_check.yml @@ -24,7 +24,7 @@ on: jobs: scala-format-check: name: Scala Format Check - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 steps: - name: Checkout code uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 @@ -33,10 +33,11 @@ jobs: fetch-depth: 0 ref: ${{ github.event.pull_request.head.ref }} - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + - name: Setup JDK 8 + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.8" + distribution: temurin + java-version: '8' - name: Run scalafmt And Print Diff continue-on-error: true diff --git a/.github/workflows/jacoco_report.yml b/.github/workflows/jacoco_report.yml index a59e7ddf9..1ba8964ff 100644 --- a/.github/workflows/jacoco_report.yml +++ b/.github/workflows/jacoco_report.yml @@ -51,7 +51,7 @@ env: jobs: detect: name: Detect Changed Files - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 timeout-minutes: 2 outputs: scala_changed: ${{ steps.filter.outputs.scala }} @@ -74,7 +74,7 @@ jobs: name: JaCoCo Report needs: detect if: needs.detect.outputs.scala_changed == 'true' - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 permissions: contents: read issues: write @@ -101,10 +101,11 @@ jobs: persist-credentials: false fetch-depth: 0 - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + - name: Setup JDK 11 + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.11.0-11" + distribution: temurin + java-version: '11' - name: Prepare testing database run: sbt flywayMigrate @@ -132,6 +133,6 @@ jobs: name: No Operation needs: detect if: needs.detect.outputs.scala_changed != 'true' - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 steps: - run: echo "No changes in Scala related files — passing." diff --git a/.github/workflows/license_check.yml b/.github/workflows/license_check.yml index b9eb16965..731482930 100644 --- a/.github/workflows/license_check.yml +++ b/.github/workflows/license_check.yml @@ -24,14 +24,15 @@ on: jobs: license-check: name: License Check - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 steps: - name: Checkout code uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 with: persist-credentials: false - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + - name: Setup JDK 8 + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.8" + distribution: temurin + java-version: '8' - run: sbt headerCheck diff --git a/.github/workflows/release_publish.yml b/.github/workflows/release_publish.yml index f1a619824..bdee402da 100644 --- a/.github/workflows/release_publish.yml +++ b/.github/workflows/release_publish.yml @@ -22,7 +22,7 @@ on: jobs: publish-to-sonatype: name: Publish to Sonatype - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 steps: - name: Checkout code uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 @@ -31,10 +31,14 @@ jobs: fetch-depth: 0 - uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + - name: Setup JDK 17 + # The Spark 4 rows of the agent are only added to the project matrix on Java 17+, so publishing has to + # run on a JDK 17 or `atum-agent-spark4_2.13` would silently never be released. The Spark 3 / model / + # reader rows still compile with `-release 8`, so their bytecode stays Java 8 compatible either way. + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.8" + distribution: temurin + java-version: '17' - name: Run sbt ci-release run: sbt ci-release @@ -47,7 +51,7 @@ jobs: publish-server-jar: name: Publish server jar - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 steps: - name: Checkout code uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 @@ -56,10 +60,12 @@ jobs: fetch-depth: 0 - uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + - name: Setup JDK 11 + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.11.0-11" + distribution: temurin + java-version: '11' + - name: Run sbt assembly (produces jar) run: sbt assembly diff --git a/.sbtrc b/.sbtrc index 61f697f85..3cbe0d7af 100644 --- a/.sbtrc +++ b/.sbtrc @@ -14,26 +14,43 @@ # limitations under the License. # -# Aliases in this file expected usage of test file naming conventions: -# - "UnitTests" suffix for test files and Suites which define unit tests -# - "IntegrationTests" suffix for test files and Suites which define integration tests -# - "CompatibilityTests" suffix for test files and Suites which define Agent <-> Server compatibility tests +# Naming convention for aliases in this file: +# - Type-based tests; `test` acts on whichever project is currently selected; no scope of their own: +# `test` - unit tests only (files/suites named `*UnitTests`) +# `testIT` - integration tests only (files/suites named `*IntegrationTests`) +# `testStandard` - both of the above combined ("standard", as opposed to the special kinds below) +# `testCompatibility` - Agent <-> Server compatibility tests (files/suites named `*CompatibilityTests`) +# +# - Scoped-based tests; `test` aliases always mean "run `testStandard` for exactly this scope": +# `testAgentSpark3` - both Spark 3 agent rows (Scala 2.12 + 2.13), recommended Java 11 runtime +# `testAgentSpark4` - the Spark 4 agent row (Scala 2.13), requires Java 17+ +# `testAgentReaderModelSpark3` - as `testAgentSpark3` but also with reader + model +# `testAgentReaderModelSpark4` - as `testAgentSpark4` but also with reader + model +# `testDB` - the `database` module +# +# - `doc` aliases mirror the `test` scopes above but generate docs instead of running tests. -# CPS QA types aliases -# Run unit tests +# Type-based tests alias test=; testOnly *UnitTests - -# Run integration tests alias testIT=; testOnly *IntegrationTests +alias testStandard=; test; testIT +alias testCompatibility=; testOnly *CompatibilityTests -# Run all standard tests (all except the special ones below) -alias testAllStandard=; test; testIT +# Scope: agent, per Spark build row +alias testAgentSpark3=; project agent_spark3; testStandard; project agent_spark32_12; testStandard +alias testAgentSpark4=; project agent_spark4; testStandard +alias docAgentSpark3=; project agent_spark3; doc; project agent_spark32_12; doc +alias docAgentSpark4=; project agent_spark4; doc -# Run specific: database tests using Balta -alias testDB=; project database; testOnly * +# Scope: reader + model + a specific agent Spark build row +alias testAgentReaderModelSpark3=; project reader; testStandard; project model; testStandard; testAgentSpark3 +alias docAgentReaderModelSpark3=; project reader; doc; project model; doc; docAgentSpark3 -# Run specific: agent-server compatibility tests -alias testCompatibility=; testOnly *CompatibilityTests +alias testAgentReaderModelSpark4=; project reader; testStandard; project model; testStandard; testAgentSpark4 +alias docAgentReaderModelSpark4=; project reader; doc; project model; doc; docAgentSpark4 + +# Scope: database +alias testDB=; project database; testStandard # Jacoco Aliases alias jacoco=; jacocoOn; jacocoAgent; jacocoReader; jacocoModel; jacocoServer; jacocoOff diff --git a/README.md b/README.md index 228a57284..dea90a5ef 100644 --- a/README.md +++ b/README.md @@ -5,16 +5,12 @@ [![Maintenance](https://img.shields.io/badge/Maintained%3F-yes-green.svg)](https://GitHub.com/Naereen/StrapDown.js/graphs/commit-activity) ![Java 11](https://img.shields.io/badge/Java_11-ED8B00?style=flat&logo=openjdk&logoColor=black) -| Atum Server | Atum Agent | Atum Model | Atum Reader | -|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| Atum Server | Atum Agent | Atum Model | Atum Reader | +|----------------------------------------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | [![GitHub release](https://img.shields.io/github/release/AbsaOSS/atum-service.svg)](https://GitHub.com/AbsaOSS/atum-service/releases/) | [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.atum-service/atum-agent-spark3_2.13/badge.svg)](https://central.sonatype.com/search?q=atum-agent&namespace=za.co.absa.atum-service) | [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.atum-service/atum-model_2.13/badge.svg)](https://central.sonatype.com/search?q=atum-model&namespace=za.co.absa.atum-service) | [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa.atum-service/atum-reader_2.13/badge.svg)](https://central.sonatype.com/search?q=atum-reader&namespace=za.co.absa.atum-service) | - - - - [Atum Service](#atum-service) - [Motivation](#motivation) - - [Features](#features) - [Modules](#modules) - [Agent `agent/`](#agent-agent) - [Reader `reader/`](#reader-reader) @@ -46,16 +42,16 @@ One of the challenges regulated industries face is the requirement to track and the accuracy and completeness of data. In an attempt to solve this data processing problem in Apache Spark applications, we propose the approach implemented in this application. -The purpose of Atum Service is to add the ability to specify "checkpoints" in Spark applications. These checkpoints -are used to designate when and what metrics are calculated to ensure that critical input values have not been modified -as well as allow for quick and efficient representation of the completeness of a dataset. This application does not -implement any checks or validations against these control measures, i.e. it does not act on them - Atum Service is, +The purpose of Atum Service is to add the ability to specify "checkpoints" in Spark applications. These checkpoints +are used to designate when and what metrics are calculated to ensure that critical input values have not been modified +as well as allow for quick and efficient representation of the completeness of a dataset. This application does not +implement any checks or validations against these control measures, i.e. it does not act on them - Atum Service is, rather, solely focused on capturing them. -The application provides a concise and dynamic way to track completeness and accuracy of data produced from source -through a pipeline of Spark applications. All metrics are calculated at a DataFrame level using various aggregation -functions and are stored on a single central place, in a relational database. Comparing control metrics for various -checkpoints is not only helpful for complying with strict regulatory frameworks, but also helps during development +The application provides a concise and dynamic way to track completeness and accuracy of data produced from source +through a pipeline of Spark applications. All metrics are calculated at a DataFrame level using various aggregation +functions and are stored on a single central place, in a relational database. Comparing control metrics for various +checkpoints is not only helpful for complying with strict regulatory frameworks, but also helps during development and debugging of your Spark-based data processing. ## Motivation @@ -64,6 +60,7 @@ Big Data strategy for a company usually includes data gathering and ingestion pr That is the definition of how data from different systems operating inside a company are gathered and stored for further analysis and reporting. An ingestion processes can involve various transformations like: + * Converting between data formats (XML, CSV, etc.) * Data type casting, for example converting XML strings to numeric values * Joining reference tables. For example this can include enriching existing @@ -79,35 +76,64 @@ the BCBS set of regulations requires analysis and reporting to be based on data Thus it is critical at the ingestion stage to preserve the accuracy and integrity of the data gathered from a source system. -The purpose of Atum is to provide means of ensuring no critical fields have been modified during the processing and no -records are added or lost. To do this the library provides an ability to calculate *control numbers* of explicitly +The purpose of Atum is to provide means of ensuring no critical fields have been modified during the processing and no +records are added or lost. To do this the library provides an ability to calculate *control numbers* of explicitly specified columns using a selection of agregate function. We call the set of such measurements at a given time -a *checkpoint* and each value - a result of the function computation - we call a *control measurement*. Checkpoints can +a *checkpoint* and each value - a result of the function computation - we call a *control measurement*. Checkpoints can be calculated anytime between Spark transformations and actions, so as at the start of the process or after its end. We assume the data for ETL are processed in a series of batch jobs. Let's call each data set for a given batch job a *batch*. All checkpoints are calculated for a specific batch. -## Features - -TBD - ## Modules ### Agent `agent/` -This module is intended to replace the current [Atum](https://github.com/AbsaOSS/atum) repository. + +This module is intended to replace the legacy [Atum](https://github.com/AbsaOSS/atum) library. It provides functionality for computing and pushing control metrics to the API located in `server/`. For more information, see the [Vocabulary section](#Vocabulary) or `agent/README.md` for more technical documentation. +#### Supported Spark / Scala / Java matrix + +The agent is cross-built per Spark major version. Pick the artifact matching the Spark your job runs on: + +| Spark | Scala | Java | Artifact | +|-------|------------|---------------------------|----------------------------------------------------| +| 3.5.x | 2.12, 2.13 | 8 minimum, 11 recommended | `atum-agent-spark3_2.12`, `atum-agent-spark3_2.13` | +| 4.0.x | 2.13 | 17+ | `atum-agent-spark4_2.13` | + +```scala +// Spark 3.5.x +libraryDependencies += "za.co.absa.atum-service" %% "atum-agent-spark3" % atumVersion + +// Spark 4.0.x +libraryDependencies += "za.co.absa.atum-service" %% "atum-agent-spark4" % atumVersion +``` + +```sh +sbt testAgentSpark3 # runs the agent test suites against both Spark 3 rows (Scala 2.12 and 2.13) +sbt testAgentSpark4 # runs the agent test suites against the Spark 4 row; needs Java 17+ +``` + +#### Logging + +The agent logs exclusively through the SLF4J API and never bundles an SLF4J binding (e.g. Logback) into its +published artifact - both `slf4j-api` and the concrete binding are supplied by the Spark runtime the agent is +deployed into (Spark always ships one). This means the agent's logging always defers to whatever logging +setup the host Spark job/cluster already has in place, rather than competing with or silently overriding it. + #### Spark 2.4 support -Because there are some java level incompatibilities between Spark 2.4 and Spark 3.x when build on Java 11+, we have to -drop support for Spark 2.4. If you need the agent to work with Spark 2.4 follow these steps: + +Because there are some Java level incompatibilities between Spark 2.4 and Spark 3.x when build on Java 11+, we dropped +support for Spark 2.4. If you need the agent to work with Spark 2.4 follow these steps: + * Switch to Java 8 * In `'build.sbt'` change the matrix rows, to be Spark 2.4 and Scala 2.11 for modules _agent_ and _model_ * Build these two modules and use them in your project ### Reader `reader/` + This module is a Scala client for applications reading metrics stored by the _Atum Service_. `FlowReader` and `PartitioningReader` resolve partitionings and retrieve checkpoint pages through the server's REST API. Applications control pagination; the Reader does not discover datasets from partial business keys or reconstruct @@ -116,32 +142,43 @@ the original parent-child lineage graph. Reader features require a compatible server version. In particular, older servers can ignore checkpoint time-window parameters; using a newer Reader alone does not enable server-side filtering. +*Scala 2.12 / 2.13, Java 8+. Has no Spark dependency, so it's usable in any JVM application regardless of Spark +version.* + ### Server `server/` -An API under construction that communicates with the Agent and with the persistent storage. It also provides measure -configuration to the agent. -The server accepts metrics potentially from several agents and saves them into database. In the future, it will be also -able to send the metrics definitions back if requested. +A REST API service offering capabilities for storing and retrieving measurement data from and to persistent storage. +The server accepts metrics potentially from several `Atum Agents` and saves them into one central relational database. +Also, it is able to send the measurements back if requested via `Atum Reader`. -Important note: the server never receives any real data - it only works with the metadata and metrics defined -by the agent! +Important note: the server never receives any real data - it only works with the metadata and metrics defined +by the agent! See `server/README.md` for more technical documentation. +*Scala 2.13, Java 11+ (deployed and CI-tested on Java 11). Has no Spark dependency.* + ### Data Model `model/` -This module defines a set of Data Transfer Objects. These are Atum-specific objects that carry data that are being +This module defines a set of Data Transfer Objects. These are Atum-specific objects that carry data that are being passed from agent to server and vice versa. +*Scala 2.12 / 2.13, Java 8+. Has no Spark dependency, so it's usable in any JVM application regardless of Spark +version.* + ### Database `database/` -This module contains a set of scripts that are used to create and maintain the database models. It also contains +This module contains a set of scripts that are used to create and maintain the database models. It also contains integration tests that are used to verify the logic of our database functions. + +*Scala 2.13, Java 11+ (deployed and CI-tested on Java 11). Has no Spark dependency.* + The database tests are integration tests in nature. Therefore, a few conditions applies: + * The tests are excluded from task `test` and are run only by a dedicated `dbTest` task (`sbt dbTest`). -* The database structures must exist on the target database - (follow the [deployment instructions in database module](database/README.md#Deployment)). -* The connection information to the DB is provided in file `database/src/test/resources/database.properties` +* The database structures must exist on the target database (follow + the [deployment instructions in database module](database/README.md#Deployment)). +* The connection information to the DB is provided in file `database/src/test/resources/database.properties` (See `database.properties.template` for syntax). ## Vocabulary @@ -151,10 +188,10 @@ This section defines a vocabulary of words and phrases used across the codebase ### Atum Agent Basically, the agent is supposed to be embedded into your application and its responsibility is to measure the -given metrics and send the results to the server. It acts as an entity responsible for spawning the `Atum Context` +given metrics and send the results to the server. It acts as an entity responsible for spawning the `Atum Context` and communicating with the server. -A user of the Atum Agent must provide certain `Partitioning` with a set of `Measures` he or she wants to calculate, +A user of the Atum Agent must provide certain `Partitioning` with a set of `Measures` he or she wants to calculate, and execute the `Checkpoint` operation. A server details are also needed to be configured. ### Partitioning @@ -164,36 +201,37 @@ want to apply particular metrics on. It's similar to data partitioning in HDFS o The order of individual `Partitions` in a given `Partitioning` matters. It's a map-like structure in which the order of keys (partition names) matters. -It's possible to define an additional metadata along with `Partitioning` - as a map-like structure with which -you can store various attributes associated with a given `Partitioning`, that you can potentially -use later in your application. Just to give you some ideas for these: +It's possible to define an additional metadata along with `Partitioning` - as a map-like structure with which +you can store various attributes associated with a given `Partitioning`, that you can potentially +use later in your application. Just to give you some ideas for these: + * a name of your application, ETL Pipeline, or your Spark job * a list of owners of your application or your dataset * source system of a given dataset * and more ### Atum Context - -This is a main entity responsible for actually performing calculations on a Spark DataFrame. Each `Atum Context` is -related to particular `Partitioning` - or to put in other words, each `Atum Context` contains all `Measures` + +This is a main entity responsible for actually performing calculations on a Spark DataFrame. Each `Atum Context` is +related to particular `Partitioning` - or to put in other words, each `Atum Context` contains all `Measures` for a specific data, defined by a given `Partitioning`, that are supposed to be calculated. ### Measure -A `Measure` defines what and how a single metric should be calculated. So it's a type of control metric to compute, -such as count, sum, or hash, that also defines a list of columns (if applicable) that should be used when actually +A `Measure` defines what and how a single metric should be calculated. So it's a type of control metric to compute, +such as count, sum, or hash, that also defines a list of columns (if applicable) that should be used when actually executing the calculation against a given Spark DataFrame. -Some `Measures` define no columns (such as `count`), some require exactly one column (such as `sum` of values for +Some `Measures` define no columns (such as `count`), some require exactly one column (such as `sum` of values for particular column), and some require more columns (such as `hash` function). ### Measurement -Practically speaking, a single `Measurement` contains a `Measure` and result associated with it. +Practically speaking, a single `Measurement` contains a `Measure` and result associated with it. ### Checkpoint -Each `Checkpoint` defines a sequence of `Measurements` (containing individual `Measures` and their results) that are +Each `Checkpoint` defines a sequence of `Measurements` (containing individual `Measures` and their results) that are associated with certain `Partitioning`. A `Checkpoint` is defined on the agent side, the server only accepts it. @@ -356,8 +394,8 @@ val currentAd = atumContext.currentAdditionalData ### Control measurement types -The control measurement of one or more columns is an aggregation function result executed over the dataset. It can be -calculated differently depending on the column's data type, on business requirements and function used. This table +The control measurement of one or more columns is an aggregation function result executed over the dataset. It can be +calculated differently depending on the column's data type, on business requirements and function used. This table represents all currently supported measurement types (aka measures): | Type | Description | @@ -373,7 +411,6 @@ represents all currently supported measurement types (aka measures): [//]: # (| controlType.absAggregatedTruncTotal | Calculates SUM(TRUNC(ABS())) of the specified column |) - ## How to generate Code coverage report - Use java version 11. @@ -394,50 +431,62 @@ The HTML and XML reports of coverage will be generated on the path: ## How to Run in IntelliJ To make this project runnable via IntelliJ, do the following: -- Make sure that your configuration in `server/src/main/resources/reference.conf` + +- Make sure that your configuration in `server/src/main/resources/reference.conf` is configured according to your needs -- When building within an IDE sure to have the option `-language:higherKinds` on in the compiler options, as it's often not picked up from the SBT project settings. +- When building within an IDE sure to have the option `-language:higherKinds` on in the compiler options, as it's often + not picked up from the SBT project settings. ## How to Run Tests ### Test controls -See the commands configured in the `.sbtrc` [(link)](https://www.scala-sbt.org/1.x/docs/Best-Practices.html#.sbtrc) file to provide different testing profiles. +See the commands configured in the `.sbtrc` [(link)](https://www.scala-sbt.org/1.x/docs/Best-Practices.html#.sbtrc) file +to provide different testing profiles. ### Run Unit Tests -Use the `test` command to execute all unit tests, skipping all other types of tests. + +Use the `test` command to execute all unit tests, skipping all other types of tests. + ```sbt sbt test ``` ### Run Integration Tests + Use the `testIT` command to execute all Integration tests, skipping all other test types. + ```sbt sbt testIT ``` ### Run All Standard Tests -Use the `testAllStandard` command to execute all unit and integration tests except for the special ones mentioned below. +Use the `testStandard` command to execute all unit and integration tests except for the special ones mentioned below. These still won't require any real DB or service to be present. + ```sbt -sbt testAllStandard +sbt testStandard ``` -### Run Special Tests +### Run Special Tests These usually have dependency on some system or service being present on the machine where the tests are being executed. These also can be performance or penetration tests. Basically the point is that these tests require special setup and are usually slower to execute. Use the `testDB` command to execute all Integration tests in `database` module, skipping all other tests and modules. + - Hint: project custom command, requiring a real DB to be present and configured + ```sbt sbt testDB ``` If you want to run all Agent <-> Server compatibility tests, use the following command. + - Hint: project custom command, requiring a real DB and service to be present and configured on the system + ```sbt sbt testCompatibility ``` diff --git a/agent/src/main/scala/za/co/absa/atum/agent/core/Logging.scala b/agent/src/main/scala/za/co/absa/atum/agent/core/Logging.scala new file mode 100644 index 000000000..894a67b67 --- /dev/null +++ b/agent/src/main/scala/za/co/absa/atum/agent/core/Logging.scala @@ -0,0 +1,40 @@ +/* + * Copyright 2021 ABSA Group Limited + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package za.co.absa.atum.agent.core + +import org.slf4j.{Logger, LoggerFactory} + +/** + * Internal logging facility of the Atum Agent. + * + * This deliberately replaces `org.apache.spark.internal.Logging`. That trait is a Spark-private API with no + * compatibility guarantees; depending on it risks `NoSuchMethodError` / `AbstractMethodError` at runtime on + * any given Spark point release, and Spark 4 in particular reworked it. slf4j is on every Spark classpath + * (Spark 3 ships slf4j 1.7.x, Spark 4 ships slf4j 2.x) and is binary stable across both, so the agent binds + * to it directly and stays portable across Spark major versions. + * + * The logger is `@transient lazy` so that mixing this trait into a serializable class (e.g. anything captured + * by a Spark closure) does not make the class unserializable. + */ +trait Logging { + + /** Name the logger is registered under. Strips the trailing `$` that Scala appends to `object` class names. */ + protected def logName: String = this.getClass.getName.stripSuffix("$") + + @transient protected lazy val log: Logger = LoggerFactory.getLogger(logName) + +} diff --git a/agent/src/main/scala/za/co/absa/atum/agent/dispatcher/ConsoleDispatcher.scala b/agent/src/main/scala/za/co/absa/atum/agent/dispatcher/ConsoleDispatcher.scala index 596142c7e..f8263eb8f 100644 --- a/agent/src/main/scala/za/co/absa/atum/agent/dispatcher/ConsoleDispatcher.scala +++ b/agent/src/main/scala/za/co/absa/atum/agent/dispatcher/ConsoleDispatcher.scala @@ -17,7 +17,7 @@ package za.co.absa.atum.agent.dispatcher import com.typesafe.config.Config -import org.apache.spark.internal.Logging +import za.co.absa.atum.agent.core.Logging import za.co.absa.atum.model.dto._ /** @@ -25,7 +25,7 @@ import za.co.absa.atum.model.dto._ */ class ConsoleDispatcher(config: Config) extends Dispatcher(config) with Logging { - logInfo("using console dispatcher") + log.info("using console dispatcher") override protected[agent] def createPartitioning(partitioning: PartitioningSubmitDTO): AtumContextDTO = { println(s"Fetching AtumContext using ConsoleDispatcher with partitioning $partitioning") diff --git a/agent/src/main/scala/za/co/absa/atum/agent/dispatcher/HttpDispatcher.scala b/agent/src/main/scala/za/co/absa/atum/agent/dispatcher/HttpDispatcher.scala index 204f72aac..39a844e9d 100644 --- a/agent/src/main/scala/za/co/absa/atum/agent/dispatcher/HttpDispatcher.scala +++ b/agent/src/main/scala/za/co/absa/atum/agent/dispatcher/HttpDispatcher.scala @@ -18,7 +18,7 @@ package za.co.absa.atum.agent.dispatcher import com.typesafe.config.Config import okhttp3.OkHttpClient -import org.apache.spark.internal.Logging +import za.co.absa.atum.agent.core.Logging import sttp.capabilities import sttp.client3._ import sttp.model.Uri diff --git a/agent/src/main/scala/za/co/absa/atum/agent/model/Measure.scala b/agent/src/main/scala/za/co/absa/atum/agent/model/Measure.scala index 4b2de2c0b..8aa8a0d35 100644 --- a/agent/src/main/scala/za/co/absa/atum/agent/model/Measure.scala +++ b/agent/src/main/scala/za/co/absa/atum/agent/model/Measure.scala @@ -99,7 +99,7 @@ object AtumMeasure { override def function: MeasurementFunction = (ds: DataFrame) => { val dataType = ds.select(measuredCol).schema.fields(0).dataType - val resultValue = ds.select(columnAggFn(castForAggregation(dataType, col(measuredCol)))).collect() + val resultValue = ds.select(columnAggFn(castForAggregation(dataType, measuredCol))).collect() MeasureResult(handleAggregationResult(dataType, resultValue(0)(0)), resultValueType) } @@ -118,7 +118,7 @@ object AtumMeasure { override def function: MeasurementFunction = (ds: DataFrame) => { val dataType = ds.select(measuredCol).schema.fields(0).dataType - val resultValue = ds.select(columnAggFn(castForAggregation(dataType, col(measuredCol)))).collect() + val resultValue = ds.select(columnAggFn(castForAggregation(dataType, measuredCol))).collect() MeasureResult(handleAggregationResult(dataType, resultValue(0)(0)), resultValueType) } @@ -139,7 +139,7 @@ object AtumMeasure { override def function: MeasurementFunction = (ds: DataFrame) => { val dataType = ds.select(measuredCol).schema.fields(0).dataType - val resultValue = ds.select(columnAggFn(castForAggregation(dataType, col(measuredCol)))).collect() + val resultValue = ds.select(columnAggFn(castForAggregation(dataType, measuredCol))).collect() MeasureResult(handleAggregationResult(dataType, resultValue(0)(0)), resultValueType) } @@ -162,7 +162,7 @@ object AtumMeasure { override def function: MeasurementFunction = (ds: DataFrame) => { val dataType = ds.select(measuredCol).schema.fields(0).dataType - val resultValue = ds.select(columnAggFn(castForAggregation(dataType, col(measuredCol)))).collect() + val resultValue = ds.select(columnAggFn(castForAggregation(dataType, measuredCol))).collect() MeasureResult(handleAggregationResult(dataType, resultValue(0)(0)), resultValueType) } @@ -196,18 +196,28 @@ object AtumMeasure { private def castForAggregation( dataType: DataType, - column: Column + columnName: String ): Column = { + val column = col(columnName) dataType match { case _: LongType => // This is protection against long overflow, e.g. Long.MaxValue = 9223372036854775807: // scala> sc.parallelize(List(Long.MaxValue, 1)).toDF.agg(sum("value")).take(1)(0)(0) // res11: Any = -9223372036854775808 - // Converting to BigDecimal fixes the issue + // Converting to BigDecimal fixes the issue. A Long always fits in decimal(38,0), so this cast can + // never fail regardless of Spark's ANSI setting - a plain `cast` is safe here. column.cast(DecimalType(38, 0)) case _: StringType => - // Support for string type aggregation - column.cast(DecimalType(38, 18)) + // Support for string type aggregation. Spark 4 defaults `spark.sql.ansi.enabled` to true, under which + // a plain `cast()` throws on a malformed numeric string instead of returning null - Spark 3's + // non-ANSI-by-default behavior this aggregation relies on (see `handleAggregationResult`'s null + // handling below). `try_cast` always returns null on a bad cast, on both Spark 3.5+ and Spark 4, + // regardless of the ANSI setting. `Column.try_cast` itself is a Spark-4-only convenience method (the + // agent's main sources are shared, unmodified, across the Spark 3 and Spark 4 build rows), so the + // version-stable SQL `try_cast` expression is used instead. + // Modifying the column name in case it contains a backtick and catering for . format. + val escapedColumnName = columnName.split("\\.", -1).map(part => s"`${part.replace("`", "``")}`").mkString(".") + expr(s"try_cast($escapedColumnName as decimal(38,18))") case _ => column } diff --git a/agent/src/main/scala/za/co/absa/atum/agent/model/MeasuresBuilder.scala b/agent/src/main/scala/za/co/absa/atum/agent/model/MeasuresBuilder.scala index 739757c23..4eff2a38c 100644 --- a/agent/src/main/scala/za/co/absa/atum/agent/model/MeasuresBuilder.scala +++ b/agent/src/main/scala/za/co/absa/atum/agent/model/MeasuresBuilder.scala @@ -16,9 +16,9 @@ package za.co.absa.atum.agent.model +import za.co.absa.atum.agent.core.Logging import za.co.absa.atum.agent.model.AtumMeasure._ import za.co.absa.atum.model.dto.MeasureDTO -import org.apache.spark.internal.Logging import scala.util.Try @@ -34,7 +34,7 @@ private [agent] object MeasuresBuilder extends Logging { case Some(value) => Some(value) case None => - logWarning(s"Measure not supported or unknown: $measure.") + log.warn(s"Measure not supported or unknown: $measure.") None } } diff --git a/agent/src/test/resources/log4j2.properties b/agent/src/test/resources/log4j2.properties new file mode 100644 index 000000000..a8588d703 --- /dev/null +++ b/agent/src/test/resources/log4j2.properties @@ -0,0 +1,20 @@ +# Copyright 2021 ABSA Group Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +rootLogger.level=ERROR +rootLogger.appenderRef.stdout.ref = console +appender.console.type = Console +appender.console.name = console +appender.console.target = SYSTEM_ERR +appender.console.layout.type = PatternLayout +appender.console.layout.pattern = %d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n diff --git a/agent/src/test/resources/logback.xml b/agent/src/test/resources/logback.xml deleted file mode 100644 index 716a839ad..000000000 --- a/agent/src/test/resources/logback.xml +++ /dev/null @@ -1,10 +0,0 @@ - - - - %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n - - - - - - diff --git a/agent/src/test/scala/za/co/absa/atum/agent/model/MeasureUnitTests.scala b/agent/src/test/scala/za/co/absa/atum/agent/model/MeasureUnitTests.scala index f00f5c299..ff04e4597 100644 --- a/agent/src/test/scala/za/co/absa/atum/agent/model/MeasureUnitTests.scala +++ b/agent/src/test/scala/za/co/absa/atum/agent/model/MeasureUnitTests.scala @@ -136,4 +136,32 @@ class MeasureUnitTests extends AnyFlatSpec with Matchers with SparkTestBase { se assert(dfFullSalaryAbsSumTruncResult.resultValueType == ResultValueType.LongValue) } + // Test for a Spark 4 behavior change: `spark.sql.ansi.enabled` defaults to `true` there, under which a plain `cast()` + // throws on a malformed numeric string instead of returning null as it does under Spark 3's non-ANSI default. + // `castForAggregation` uses `try_cast` instead, which always returns null on a bad cast regardless of the ANSI + // setting - this must hold on both Spark versions, so the test forces ANSI on explicitly to also cover Spark 3 + // defensively. + it should "not throw and should treat a malformed numeric string as null when aggregating, with ANSI enabled" in { + val salarySum = SumOfValuesOfColumn("salary") + + val dfWithBadSalary = spark + .createDataFrame( + Seq( + ("id", "firstName", "lastName", "email", "email2", "profession", "1000.50"), + ("id", "firstName", "lastName", "email", "email2", "profession", "not-a-number") + ) + ) + .toDF("id", "firstName", "lastName", "email", "email2", "profession", "salary") + + val originalAnsiSetting = spark.conf.get("spark.sql.ansi.enabled", "false") + spark.conf.set("spark.sql.ansi.enabled", "true") + try { + val result = salarySum.function(dfWithBadSalary) + result.resultValue shouldBe "1000.5" + result.resultValueType shouldBe ResultValueType.BigDecimalValue + } finally { + spark.conf.set("spark.sql.ansi.enabled", originalAnsiSetting) + } + } + } diff --git a/build.sbt b/build.sbt index e8d360ef5..c99b44b4f 100644 --- a/build.sbt +++ b/build.sbt @@ -17,7 +17,7 @@ import sbt.* import sbt.Keys.* import Dependencies.* -import Dependencies.Versions.spark3 +import Dependencies.Versions.{spark3, spark4} import VersionAxes.* ThisBuild / scalaVersion := Setup.scala213.asString @@ -31,14 +31,25 @@ val limitedProject: Boolean = Setup.currentJava < Setup.recommendedJava initialize := { val _ = initialize.value // Ensure previous initializations are run - assert(Setup.currentJava >= Setup.requiredJava, - s"Running on Java version ${Setup.currentJava}, required is at least version ${Setup.requiredJava}, recommended is ${Setup.recommendedJava}") + assert( + Setup.currentJava >= Setup.spark3RequiredJava, + s"Running on Java version ${Setup.currentJava}, required is at least version ${Setup.spark3RequiredJava}, recommended is ${Setup.recommendedJava}" + ) if (limitedProject) { val log = Keys.sLog.value - log.warn(s"Some nodules will not be loaded, because they require at least Java ${Setup.recommendedJava} while Java ${Setup.currentJava} has been found") + log.warn( + s"Some nodules will not be loaded, because they require at least Java ${Setup.recommendedJava} while Java ${Setup.currentJava} has been found" + ) log.warn("""Affected modules are: "atum-server", "atum-database"""") } + + if (!Setup.spark4Supported) { + val log = Keys.sLog.value + log.warn( + s"The Spark ${Dependencies.Versions.spark4} rows of the 'atum-agent' module will not be loaded, because they require at least Java ${Setup.spark4RequiredJava} while Java ${Setup.currentJava} has been found" + ) + } } enablePlugins(FlywayPlugin) @@ -70,7 +81,7 @@ lazy val server = { version := git.gitDescribedVersion.value.map(_.takeWhile(_ != '-')).getOrElse("unknown"), buildInfoKeys := Seq[BuildInfoKey]( version, - "fullVersion" -> git.gitDescribedVersion.value.getOrElse("unknown"), + "fullVersion" -> git.gitDescribedVersion.value.getOrElse("unknown") ), buildInfoPackage := "za.co.absa.atum.server.api.common.http" ): _* @@ -90,18 +101,43 @@ lazy val server = { /** * Module `agent` is the library to be plugged into the Spark application to measure the data and send it to the server + * + * It is cross-built per Spark major version, each axis carrying its own Scala and Java baseline: + * - Spark 3.5.x -> Scala 2.12 + 2.13, Java 8 (`atum-agent-spark3_2.12`, `atum-agent-spark3_2.13`) + * - Spark 4.0.x -> Scala 2.13, Java 17 (`atum-agent-spark4_2.13`) + * + * The Spark 4 rows require a JDK 17+ to compile, so they are only added when sbt itself runs on one. */ -lazy val agent = (projectMatrix in file("agent")) - .disablePlugins(sbtassembly.AssemblyPlugin) - .settings( - Setup.commonSettings ++ Seq( - name := "atum-agent", - javacOptions ++= Setup.clientJavacOptions - ): _* - ) - .addSparkCrossBuild(SparkVersionAxis(spark3), Setup.clientSupportedScalaVersions, Dependencies.agentDependencies) - .dependsOn(model) - .enablePlugins(JacocoFilterPlugin) +lazy val agent = { + val agent = (projectMatrix in file("agent")) + .disablePlugins(sbtassembly.AssemblyPlugin) + .settings( + Setup.commonSettings ++ Seq( + name := "atum-agent" + ): _* + ) + .addSparkCrossBuild( + SparkVersionAxis(spark3), + Setup.clientSupportedScalaVersions(spark3), + Dependencies.agentDependencies + ) + + // Appending Spark 4 if needed (based on Java requirements) while also building and supporting Spark 3 on newer + // Java (that's why append, otherwise Spark 3 would be skipped on newer Java). + val agentWithSpark = if (Setup.spark4Supported) { + agent.addSparkCrossBuild( + SparkVersionAxis(spark4), + Setup.clientSupportedScalaVersions(spark4), + Dependencies.agentDependencies + ) + } else { + agent + } + + agentWithSpark + .dependsOn(model) + .enablePlugins(JacocoFilterPlugin) +} /** * Module `model` is the data model for data exchange with server @@ -111,7 +147,7 @@ lazy val model = (projectMatrix in file("model")) .settings( Setup.commonSettings ++ Seq( name := "atum-model", - javacOptions ++= Setup.clientJavacOptions, + javacOptions ++= Setup.clientJavacOptions() ): _* ) .addScalaCrossBuild(Setup.clientSupportedScalaVersions, Dependencies.modelDependencies) @@ -147,7 +183,7 @@ lazy val reader = (projectMatrix in file("reader")) .settings( Setup.commonSettings ++ Seq( name := "atum-reader", - javacOptions ++= Setup.clientJavacOptions + javacOptions ++= Setup.clientJavacOptions() ): _* ) .addScalaCrossBuild(Setup.clientSupportedScalaVersions, Dependencies.readerDependencies) diff --git a/project/Dependencies.scala b/project/Dependencies.scala index f4cbf18c1..829131c59 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -20,6 +20,7 @@ object Dependencies { object Versions { val spark3 = "3.5.5" + val spark4 = "4.1.2" val scalatest = "3.2.15" val scalaMockito = "1.17.12" @@ -29,7 +30,8 @@ object Dependencies { val specs2 = "4.10.0" val typesafeConfig = "1.4.2" - val sparkCommons = "0.6.3" + val sparkCommonsSpark3 = "0.6.3" + val sparkCommonsSpark4 = "1.0.0" val sttpClient = "3.5.2" //last supported version for Java 8 val sttpCirceJson = "3.9.7" @@ -38,8 +40,6 @@ object Dependencies { val fadb = "0.7.0" - val logback = "1.2.3" - val zio = "2.0.19" val zioLogging = "2.2.0" val logbackZio = "1.4.7" @@ -70,6 +70,14 @@ object Dependencies { def getVersionUpToMajor(version: String): String = { truncateVersion(version, 1) } + + /** + * Spark 4 is the cut-off for several breaking changes (e.g. Java 17 baseline, Scala 2.12 dropped). + * Build settings key off this predicate rather than off an exact version. + */ + def isSpark4OrLater(sparkVersion: String): Boolean = { + getVersionUpToMajor(sparkVersion).toInt >= 4 + } } private def testDependencies: Seq[ModuleID] = { @@ -185,21 +193,20 @@ object Dependencies { } def agentDependencies(sparkVersion: String, scalaVersion: Version): Seq[ModuleID] = { - val sparkMinorVersion = Versions.getVersionUpToMinor(sparkVersion) val scalaMinorVersion = Versions.getVersionUpToMinor(scalaVersion.asString) + val spark4OrLater = Versions.isSpark4OrLater(sparkVersion) + + val sparkCommonsVersion = if (spark4OrLater) Versions.sparkCommonsSpark4 else Versions.sparkCommonsSpark3 lazy val sparkCore = "org.apache.spark" %% "spark-core" % sparkVersion % Provided lazy val sparkSql = "org.apache.spark" %% "spark-sql" % sparkVersion % Provided lazy val typeSafeConfig = "com.typesafe" % "config" % Versions.typesafeConfig - lazy val sparkCommons = "za.co.absa" % s"spark-commons-spark${sparkMinorVersion}_$scalaMinorVersion" % Versions.sparkCommons - lazy val sparkCommonsTest = "za.co.absa" % s"spark-commons-test_$scalaMinorVersion" % Versions.sparkCommons % Test + lazy val sparkCommonsTest = "za.co.absa" % s"spark-commons-test_$scalaMinorVersion" % sparkCommonsVersion % Test lazy val sttpClient3 = "com.softwaremill.sttp.client3" %% "core" % Versions.sttpClient lazy val sttpOkHttpBackend = "com.softwaremill.sttp.client3" %% "okhttp-backend" % Versions.sttpClient - lazy val logback = "ch.qos.logback" % "logback-classic" % Versions.logback - lazy val nameOf = "com.github.dwickern" %% "scala-nameof" % Versions.scalaNameof % Provided // it's provided, as it's a macro needed only at runtime lazy val balta = "za.co.absa" %% "balta" % Versions.balta % Test @@ -208,11 +215,9 @@ object Dependencies { sparkCore, sparkSql, typeSafeConfig, - sparkCommons, sparkCommonsTest, sttpClient3, sttpOkHttpBackend, - logback, nameOf ) ++ testDependencies :+ balta diff --git a/project/Setup.scala b/project/Setup.scala index 0dff42a3e..5631ba75f 100644 --- a/project/Setup.scala +++ b/project/Setup.scala @@ -23,15 +23,19 @@ import za.co.absa.commons.version.Version object Setup { + val currentJava: Double = sys.props("java.specification.version").toDouble + //supported Java versions - val requiredJava: Double = "1.8".toDouble + val spark3RequiredJava: Double = "1.8".toDouble // absolute minimum val recommendedJava: Double = "11".toDouble - val currentJava: Double = sys.props("java.specification.version").toDouble + + //Spark 4 has a Java 17 baseline, so its build rows cannot even be compiled on an older JDK + val spark4RequiredJava: Double = "17".toDouble + val spark4Supported: Boolean = currentJava >= spark4RequiredJava //possible supported Scala versions - val scala211: Version = Version.asSemVer("2.11.12") val scala212: Version = Version.asSemVer("2.12.18") - val scala213: Version = Version.asSemVer("2.13.13") + val scala213: Version = Version.asSemVer("2.13.18") lazy val commonSettings: Seq[SettingsDefinition] = Seq( scalacOptions ++= Setup.commonScalacOptions, @@ -46,6 +50,25 @@ object Setup { scala213, ) + /** + * Scala versions supported for a given Spark axis. Spark 4 dropped Scala 2.12, so the 2.12 row must never be + * attempted against it - resolution of `spark-core_2.12:4.x` would simply fail. + */ + def clientSupportedScalaVersions(sparkVersion: String): Seq[Version] = { + if (Dependencies.Versions.isSpark4OrLater(sparkVersion)) Seq(scala213) + else clientSupportedScalaVersions + } + + //Java bytecode level the client modules (agent, model, reader) are compiled down to + val spark3ClientJavaTarget: String = "1.8" + val spark4ClientJavaTarget: String = "17" + + /** Spark 3 artifacts stay on Java 8 bytecode; Spark 4 requires a Java 17 baseline. */ + def clientJavaTarget(sparkVersion: String): String = { + if (Dependencies.Versions.isSpark4OrLater(sparkVersion)) spark4ClientJavaTarget + else spark3ClientJavaTarget + } + val commonScalacOptions: Seq[String] = Seq( "-unchecked", "-deprecation", @@ -63,19 +86,23 @@ object Setup { "-Ymacro-annotations" ) - val clientJavacOptions: Seq[String] = Seq("-source", "1.8", "-target", "1.8", "-Xlint") - def clientScalacOptions(scalaVersion: Version): Seq[String] = { + def clientJavacOptions(javaTarget: String = spark3ClientJavaTarget): Seq[String] = + Seq("-source", javaTarget, "-target", javaTarget, "-Xlint") + + def clientScalacOptions(scalaVersion: Version, javaTarget: String = spark3ClientJavaTarget): Seq[String] = { + //scalac's `-release` takes the JEP-322 feature number, so "1.8" has to be normalised to "8" + val release = javaTarget.stripPrefix("1.") if (scalaVersion >= scala213) { Seq( - "-release", "8", + "-release", release, "-language:higherKinds", "-Ymacro-annotations" ) } else { Seq( - "-release", "8", + "-release", release, "-language:higherKinds", - "-target:8" + s"-target:$release" ) } } diff --git a/project/VersionAxes.scala b/project/VersionAxes.scala index 27e3dbd56..9a3a4d03b 100644 --- a/project/VersionAxes.scala +++ b/project/VersionAxes.scala @@ -43,15 +43,28 @@ object VersionAxes { dependenciesFnc: (String, Version) => Seq[ModuleID], settings: Def.SettingsDefinition*): ProjectMatrix = { val sparkVersion = sparkAxis.sparkVersion - scalaVersions.foldLeft(projectMatrix) { case (currentProjectMatrix, scalaVersion) => + val javaTarget = Setup.clientJavaTarget(sparkVersion) + //guard: never build a Scala version the Spark axis does not support (notably 2.12 against Spark 4), + //otherwise dependency resolution blows up on a non-existent `spark-core_2.12:4.x`. Fails loudly instead + //of silently dropping the unsupported row, so a misconfigured call site is caught at build-definition time. + val supportedScalaVersions = Setup.clientSupportedScalaVersions(sparkVersion) + val unsupportedScalaVersions = scalaVersions.filterNot(supportedScalaVersions.contains) + require( + unsupportedScalaVersions.isEmpty, + s"Spark $sparkVersion does not support Scala version(s) ${unsupportedScalaVersions.map(_.asString).mkString(", ")}. " + + s"Supported Scala versions for this Spark axis are: ${supportedScalaVersions.map(_.asString).mkString(", ")}." + ) + + scalaVersions.distinct.foldLeft(projectMatrix) { case (currentProjectMatrix, scalaVersion) => currentProjectMatrix.customRow( scalaVersions = Seq(scalaVersion.asString), axisValues = Seq(sparkAxis, VirtualAxis.jvm), _.settings( moduleName := camelCaseToLowerDashCase(name.value + sparkAxis.directorySuffix), - scalacOptions ++= Setup.clientScalacOptions(scalaVersion), + scalacOptions ++= Setup.clientScalacOptions(scalaVersion, javaTarget), + javacOptions ++= Setup.clientJavacOptions(javaTarget), libraryDependencies ++= dependenciesFnc(sparkVersion, scalaVersion), - printVersionInfo := streams.value.log.info(s"Building ${name.value} with Spark $sparkVersion, Scala ${scalaVersion.asString}"), + printVersionInfo := streams.value.log.info(s"Building ${name.value} with Spark $sparkVersion, Scala ${scalaVersion.asString}, Java $javaTarget"), (Compile / compile) := ((Compile / compile) dependsOn printVersionInfo).value, ).settings(settings *) )