I have created project pyspark-sql-conda-env: https://gitlab.wikimedia.org/repos/data-engineering/pyspark-sql-conda-env

The idea is to have a simple pyspark entry point environment with more recent Spark and Iceberg (say 3.3.2 and 1.61), and this entry point should be usable with arbitrary SQL, although the first use case would be to do Sql CALLs to call Iceberg procedures.

From Airflow we would call it like:

sql_args = [
     "--sql",
                         f"""
                         CALL spark_catalog.system.expire_snapshots(
                             table => 't1',
                             older_than =>
                               TIMESTAMP '123',
                             max_concurrent_deletes => 5,
                             stream_results => true
                         )
                         """,
 ]

 process_events = SparkSubmitOperator.for_virtualenv(
     task_id="spark_run_sql",
     virtualenv_archive=props.conda_env,
     entry_point="bin/run_sql.py",
     driver_memory=props.driver_memory,
     driver_cores=props.driver_cores,
     executor_memory=props.executor_memory,
     executor_cores=props.executor_cores,
     conf=common_spark_conf,
     launcher="skein",
     application_args=sql_args,
     use_virtualenv_spark=True,
     default_env_vars={
         "SPARK_HOME": "venv/lib/python3.10/site-packages/pyspark",  # point to the packaged Spark
         "SPARK_CONF_DIR": "/etc/spark3/conf",
     },
    ...
     spark_datahub_lineage_enabled=False,  # Iceberg tables not supported
 )

Any SQL parameter building will have to be done upstream. Thus we could very well hide this whole SparkSubmitOperator.for_virtualenv() call on Airflow side with a, say, PySparkSqlOperator() oeprator. Nice, but not neccesary.

@xcollazo hm! Cool!

Q: Is pyspark needed to run Spark SQL?

You want to use custom spark versions, right? That indeed means you'd need a custom conda env (with the correct 'pyspark' Spark version installed), but do you need / want to launch with custom pyspark stuff?

Could you just launch with bin/spark-sql CLI and avoid any need for custom python code?

but do you need / want to launch with custom pyspark stuff?
Could you just launch with bin/spark-sql CLI and avoid any need for custom python code?

That is what I have been pondering as well. To avoid any potential semantic changes, I wanted to mimic as close as what happens today, which includes supporting anything that our SparkSqlOperator supports. Over there, we launch via our org.apache.spark.sql.hive.thriftserver.WMFSparkSQLCLIDriver which is via spark-submit, and that particular one supports https files. I haven't tested it yet, but using this will open us up to incompatibilities between that file and newer Spark versions, etc. So it gets complicated.

So agreed that just calling bin/spark-sql on the venv would be easiest, but then it won't support https files I think.

Ah! TIL about that feature! Okay!

Then the next question is: Maybe you just need bin/spark-submit with --jars wmf-sparksqlclidriver.jar?

Could you achieve this by adding for_virtualenv support to our SparkSqlOperator? Allowing you to choose the conda env it uses to launch the SQL?

Could you achieve this by adding for_virtualenv support to our SparkSqlOperator? Allowing you to choose the conda env it uses to launch the SQL?

I did something in those lines, but scoped to only support Spark 3.3.2 for now. Will send draft patch in momentarily, kindly please review if you have time.

xcollazo opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/1226

Draft: analytics: Intro SparkSql332Operator. Do Iceberg table maintenance with Spark 3.3.2.

The current Spark 3.1.2 compiled WMFSparkSQLCLIDriver is indeed incompatible with Spark 3.3.2:

Exception in thread "main" java.lang.NoSuchMethodError: org.apache.spark.sql.internal.SharedState$.resolveWarehousePath(Lorg/apache/spark/SparkConf;Lorg/apache/hadoop/conf/Configuration;Lscala/collection/Map;)Lscala/collection/Map;
	at org.apache.spark.sql.hive.thriftserver.WMFSparkSQLCLIDriver$.main(WMFSparkSQLCLIDriver.scala:142)
	at org.apache.spark.sql.hive.thriftserver.WMFSparkSQLCLIDriver.main(WMFSparkSQLCLIDriver.scala)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
	at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:958)
	at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:180)
	at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:203)
	at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:90)
	at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:1046)
	at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1055)
	at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

Argh.

xcollazo opened https://gitlab.wikimedia.org/repos/data-engineering/patches/wmf-sparksqlclidriver/-/merge_requests/2

Compile WMFSparkSQLCLIDriver against Spark 3.3.2.

In T391280#10731267, @CodeReviewBot wrote:

xcollazo opened https://gitlab.wikimedia.org/repos/data-engineering/patches/wmf-sparksqlclidriver/-/merge_requests/2

Compile WMFSparkSQLCLIDriver against Spark 3.3.2.

This works! 🎉

Now let's clean things up...

xcollazo merged https://gitlab.wikimedia.org/repos/data-engineering/patches/wmf-sparksqlclidriver/-/merge_requests/2

Compile WMFSparkSQLCLIDriver against Spark 3.3.2.

Updated README: https://gitlab.wikimedia.org/repos/data-engineering/patches/wmf-sparksqlclidriver

Deployed WMFSparkSQLCLIDriver artifact via mvn deploy to https://archiva.wikimedia.org/repository/releases/org/wikimedia/spark/wmf-sparksqlclidriver-3_3_2/2.0.0/

From MR 1226:

(This MR pulls artifacts from https://gitlab.wikimedia.org/repos/data-engineering/patches/wmf-sparksqlclidriver and https://gitlab.wikimedia.org/repos/data-engineering/patches/pyspark-sql-conda-env).

In this MR we:

• Introduce a SparkSql332Operator that inherits from SparkSqlOperator but under the cover uses Spark 3.3.2 (compared to production's 3.1.2) via a conda environment. This operator tries to hide the complexities of conda environments away.
• Refactors Iceberg table maintenance tasks to use the above.
  
  Why a SparkSql332Operator operator? Why not just add a parameter to SparkSqlOperator?
  
  The configuration necessary for SparkSql332Operator is very specific to the use of a conda enviornment, the settings to pickup Spark 3.3.2, and the fact that this operator may be temporary till we upgrade the cluster. An inheritance relationship seems more appropriate, and keeps SparkSqlOperator clean.

MR 1226 is a bit different now after a couple reviews:

(This MR pulls artifacts from https://gitlab.wikimedia.org/repos/data-engineering/patches/wmf-sparksqlclidriver and https://gitlab.wikimedia.org/repos/data-engineering/patches/pyspark-sql-conda-env).

In this MR we:

• Introduce a spark_version: Literal["3.1.2", "3.3.2"] parameter on SparkSqlOperator to be able to run Spark 3.3.2 via a conda environment on top of YARN. This mechanism tries to hide the complexities of conda environments away by configuring all 3.3.2 bits automatically. We default to the usual Spark 3.1.2 run directly over YARN otherwise.
• Refactors Iceberg table maintenance tasks to use the above.

xcollazo merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/1226

analytics: Allow SparkSqlOperator to run Spark 3.3.2. Do Iceberg table maintenance with Spark 3.3.2.

Mentioned in SAL (#wikimedia-operations) [2025-04-15T17:23:40Z] <xcollazo@deploy1003> Started deploy [airflow-dags/analytics@f650091]: Pickup latest artifacts. T391280.

Mentioned in SAL (#wikimedia-operations) [2025-04-15T17:24:23Z] <xcollazo@deploy1003> Finished deploy [airflow-dags/analytics@f650091]: Pickup latest artifacts. T391280. (duration: 01m 08s)

Running an Iceberg table maintenance and things look good.