Extract and send events with raw sql from spark.

[Imbruced]

Problem

Currently for spark application we don't send any information about used sql query, for example
spark.sql(
"SELECT id FROM table"
).save......

will track that we used table and saved that somewhere but the raw sql won't be recorded in any event

"SELECT id FROM table"

structure.

Closes: #ISSUE-NUMBER

Solution

We can look into QuerExecution children up we find the sqlQuery, the downside of this using bfs algorithm is we keep only one query. So in case like

spark.sql("SELECT 1 as id").createOrReplaceTempView("tab1")
spark.sql("SELECT 2 as id").createOrReplaceTempView("tab2")

spark.sql("select * from tab1 union all select * from tab2")

we store only query select * from tab1 union all select * from tab2

that's the limitation of sql query facet where we can store only one value.

Field which we are looking for is available only for spark 3.3 and above

One-line summary:

Traversing QueryExecution to get sql field with bfs algorithm.

Checklist

• You've signed-off your work
• Your pull request title follows our guidelines
• Your changes are accompanied by tests (if relevant)
• Your change contains a small diff and is self-contained
• You've updated any relevant documentation (if relevant)
• Your comment includes a one-liner for the changelog about the specific purpose of the change (not required for changes to tests, docs, or CI config)
• You've versioned the core OpenLineage model or facets according to SchemaVer (if relevant)
• You've added a header to source files (if relevant)

---

SPDX-License-Identifier: Apache-2.0
Copyright 2018-2024 contributors to the OpenLineage project

[dolfinus]

Please update PR description and title

[mobuchowski]

Just did only high level review with two items - because IMO we have to rethink those before moving forward with the PR.

[mobuchowski]

I don't think it makes sense to put that here.

First, there can be multiple Spark SQL queries tied to a single "application" - script - think just

df = spark.sql("select * from a");
df2 = spark.sql("select * from b");
df.join(df2, ...).write() ...

In this case, you'd get two events here with application run id, without ability to tie them to particular Spark Actions.

I think the better idea is to check if we can get something like executionId from Spark Session, and store the SQL in some map keyed by this executionId. Then, attach the SqlJobFacet to the next (all) COMPLETE and RUNNING events send with that executionId.