Five highlights on the Spark 3.0.0 Release

Spark 3.0.0 was officially released yesterday (18/Jun/2020), and it is a major change (no pun intended) in the most popular data processing engine of the world (Flink, you are great as well, but Spark is a bit more famous).

First, a reminder, what actually is Spark? It is an open-source, fault-tolerant, and high-speed in-memory data processing engine. It supports batch and streaming processing plus some cool analytics. The ML part was not much used in Spark 2, but now we may have a shift with the new change in Spark 3!

Besides Databricks services and your Spark running on-premises setup, this will affect lots of workloads. Cloudera distributions, companies that built products on top of Spark and every public cloud provider (they all have a Spark managed service) will react to it, and I’m eagerly waiting to see what they will build! (Databricks already made Spark 3.0 available!).

Going back a few years, Spark 2 brought many fundamental new things to the table (the catalyst optimizer is the most remarkable one, in my opinion) that really pushed Spark to the enterprise world. The integration with different input sources, the Hive metastore connection, Structured Streaming and much more turned Spark into what it is today: the most used engine for building data pipelines and workloads on top of data lakes. So, let’s cut to the chase, Thiago? What do you think are the essential things in the earth-shattering release?

1. SQL ANSI and compatibility

If Excel rules the world, SQL rules the underworld of data. It is easy to learn, it is intuitive, and you can do a lot with it (and see some nice SQL abuses on this Reddit post). However, we always had to learn some tips and tricks for writing SQL in different systems because they had minor distinctions. The SQL ANSI sets up a standard for it that Spark was somewhat behind — I always made mistakes in filter for aggregate expressions, but that now is fixed!

Of course, we are not completely there as we are not full ANSI/SQL with Spark 3.0, but you can check the status of that endeavor here.

Does that mean that if I upgrade my Spark version, my old Spark-SQL code will stop working? Well, upgrading major versions can always be a problem, but the ANSI SQL part can be toggled as a configuration to reserve some keywords by just adding:

spark.sql.ansi.enabled = True

2. Native Prometheus

Observability on your jobs is not a nice-to-have, but a MUST, when running production workloads. Even though we could get lots of metrics from Spark jobs, I think almost everyone built some custom-made gateways to forward their metrics to somewhere else to monitor and alarm about your jobs. Especially JVM memory and Garbage Collection information, right? :-)

Monitoring and alerting nowadays is almost synonymous with Prometheus. This amazing open-source project by CNCF is excellent, with lots of contributors, fast-moving, and easy to set up. The majority of Spark 2.x projects we’ve worked we would build a Prometheus exporter to Graphite and scrape it so we could get metrics information and enable alerting for the executors. How exactly?

2.1 Spark 2 Prometheus Monitoring

(I) Exporting data into Graphite.
We would deploy the exporter in a container and run in parallel to the cluster with proper auto-scaling. A simple docker-compose file for local running and its configuration file can be seen below (the configuration files is the propper tagging of metrics so we can query them with Prometheus).

(II) When running your spark application, you must have a proper metrics.properties file on your $SPARK_HOME/conf/ folder that points to the Graphite Exporter from (I) as the sink.

As you can see, not that complex but cumbersome. Especially when you start to have hundreds of new Spark apps, and some people may forget this — and you may just realize that when the app was not running for a few hours. Of course, you can add these steps to your CD pipeline, but you can imagine how much work goes into that automation.

2.1 On Spark 3.0? Much easier…

Besides exposing Executor metrics via the REST API on/applications/[app-id]/executors we will now have a native Prometheus endpoint /metrics/executors/prometheus, where you only need to toggle the configuration spark.ui.prometheus.enabled.

THAT. IS. IT. A toggle… :-D

3. Languages version change, Hadoop and Kubernetes

There are 4 SDKs for Spark: Python, Scala, Java, and R (the .Net one is not official). The Python one is the most used, and that is why Spark 3 pays special attention to Pandas integrations, UDF optimization on Python, and more. And, as expected in such a major version changes, some things will be added, and others deprecated. So these are the changes:

• Python 2 support is deprecated, and I couldn’t be happier!
• R versions older than 3.4 are also deprecated.
• Java 11 Support.
• Hadoop 3 support and remove Hadoop 2.6 support.
• Scala 2.12 from now on — and this create an ambiguity with DataStreamWritter.foreachBatch, so be careful.
• Kubernetes is becoming a reliable way of running Spark apps. With the new shuffle and Kerberos support for k8s, we will start seeing enterprises doing this without the fear of “being first”.

4. Dynamic Partition Pruning

Take a look at the original ticket here and check the last PR here. If I could explain this in 2 paragraphs, I would, but I cannot. So I refer you to this excellent presentation we had at Spark + AI Summit last year in Amsterdam. Kudos to @BogdanGhit and Juliusz Sompolski.

So, why is this important and amazing? Querying star-schema tables with small dimensions was not very performative, and we had to change the data structure before doing it. Now, bam! Out-of-the-box fast queries for this case that is so crucial do advanced analytics (also know as the TPC-DS data type).

5. Adaptive Query Execution

Spark SQL is being used more and more these last years with a lot of effort targeting the SQL query optimizer, so we have the best query execution plan. When you write a SQL query for Spark with your language of choice, Spark takes this query and translates it into a digestible form (logical plan). During this phase, it optimizes the logical plan by applying a set of rule-based transformations (constant folding, filter push-down, column pruning, and others). The next step is the physical planning phase, where it generates an executable plan (physical plan) that tells how the computation is distributed amongst the cluster.

Based on this understanding, you can imagine how important it is to choose a better plan for running your query. This involves gathering data statistics and cost-based optimization. For example, should I join via sort merge or via broadcast hash? What is the order of join if I am doing a multi-way join? If you feed the optimizer with outdated statistics, you will get a non-optimal plan and poorly execute your query. So now we can reoptimize and adjust the query plan based on statistics collected during the query execution! Magic, right?

If you want to understand the fundamental changes in this, similar to the Dynamic Pruning, I would have to write an entire article just on it. So I can tell you to look at the original epic here, take a look at the brilliant @maryannxue code on the “Dynamically reuse subqueries” PR here and go through this somewhat older session (2018) at the Spark Summit by Carson Wang.

Conclusion

Spark continues to change the world of Data with this new release. And it leaves a pertinent message, like it or not:

Python, Advanced Analytics, SQL in Data Warehouses, Data Streaming and Deep Learning are the focus now.

Do you want to contribute to Spark? Review some PR's? Help with the documentation? We all need you there, so take a look at the Issues page here and contribute!