Spark Operator

I've found that managing Spark clusters in Kubernetes is far easier using the Spark Operator. Here are some commands that helped me diagnose issues.

Dude, where's my appliction?

List your Spark applications with:

kubectl get sparkapplications

or

kubectl get sparkapplications spark-connector -o yaml

to see what might be causing trouble for, say, the connector.

It can be annoying when you can't delete a sparkapplication with

kubectl delete sparkapplication YOUR_APP

even though it's running. In my case, I thought a 

kubectl rollout restart deployment spark-kubernetes-operator

left an orphaned cluster.

It's possible that you don't see anything even though there are Spark pods clearly there. In this case:

kubectl describe pod POD_NAME

and you should see something like:

...

Controlled By:  StatefulSet/XXX

...

Great, so it looks like the Spark Operator has set the cluster up by delegating to Kubernetes primitives. Let's see them:

kubectl get statefulsets

and then we can just:

kubectl delete statefulset XXX

OK, so, dude, where's my cluster

But we're barking up the wrong tree. The YAML to create a cluster has kind: SparkCluster so we're using the wrong CRD with sparkapplications.

kubectl get crd | grep spark

sparkclusters.spark.apache.org                              2025-11-04T10:52:56Z

...

Right, so now:

kubectl delete sparkclusters YOUR_CLUSTER

Python

As a little aside, I was seeing strange errors when running PySpark commands that appeared to be a versioning problems. A few commands that came in useful were:

import sys

print(sys.path)

to print where the Python executable was getting its libraries from and:

from pyspark.version import __version__

print(__version__)

to make sure we really did have the correct PySpark version. 

As it happened, it was the wrong version of the Iceberg runtime in spark.jars.packages.

Spark and K8s on AWS

This is a high-level overview of my work creating a Spark cluster running in a managed AWS Kubernetes cluster (EKS) giving Spark the permissions to write to cross cloud storage. I might write further 

EKS

Create a Kubernetes cluster with:

eksctl create cluster --name spark-cluster --nodes 3

Note that this can take about 15 minutes.

Note that you can have several K8s contexts on your laptop. You can see them with:

kubectl config get-contexts

and choose one with:

kubectl config use-context <your-cluster-context-name>

then you can run kubectl get pods -A and see K8s in the cloud.

A couple of caveats: I had to install aws-iam-authenticator by building it from scratch. This also required me to install GoLang. Binaries are installed in ~/go/bin.

Authorization

The hard thing about setting up a Kubernetes cluster in AWS is configuring permissions. This guide helped me but it was strictly the section Option 2: Using EKS Pod Identity that helped me. 

Basically, you have to configure a bridge between the K8s cluster and AWS. This is done through through a CSI (Container Storage Interface), a standard K8s mechanism for storage but in this case it's a means for storing secrets.

The high-level recipe for using secrets is:

1. Create the secret with aws secretsmanager create-secret ... 
2. Create a policy with aws iam create-policy... This reference the secret's ARNs in step 1.
3. Create a role with aws iam create-role... that allows a role to be assumed via STS.
4. Attach the policy in step 2 with the role in step 3 with aws iam attach-role-policy... 
5. Create a SecretProviderClass with kubectl apply -f...  that references the secrets created in step 1
6. Associate your K8s Deployment with the SecretProviderClass using its volumes.

Thanks to the installation of the CSI addons, secrets can be mounted in a Polaris container that are then used to vend credentials to Spark.

Diagnosing Spark in the Cluster

Connecting to Spark from my laptop produced these logs:

25/09/23 18:15:58 INFO StandaloneAppClient$ClientEndpoint: Connecting to master spark://a457XXX.eu-west-2.elb.amazonaws.com:7077...

25/09/23 18:15:58 INFO TransportClientFactory: Successfully created connection to a457XXX.eu-west-2.elb.amazonaws.com/3.9.78.227:7077 after 26 ms (0 ms spent in bootstraps)

25/09/23 18:16:18 INFO StandaloneAppClient$ClientEndpoint: Connecting to master spark://a457XXX.eu-west-2.elb.amazonaws.com:7077...

25/09/23 18:16:38 INFO StandaloneAppClient$ClientEndpoint: Connecting to master spark://a457XXX.eu-west-2.elb.amazonaws.com:7077...

where the connections were timing out. So I logged in with:

kubectl exec -it prod-master-0 -- /bin/bash

and looked at the actual logs under /opt/spark/ which were far more illuminating.

Spark Kubernetes Operator

I used the Spark operator to configure a Spark cluster for me. However, examples/prod-cluster-with-three-workers.yaml seemed to be out of synch with the CRDs installed by Helm(?). The apiVersion seems to need to be:

apiVersion: spark.apache.org/v1alpha1

This change meant I could then start my Spark cluster.

Aside: to restart the whole cluster, use:

kubectl delete -f spark_cluster.yaml

kubectl apply  -f spark_cluster.yaml

AWS CLI annoyances

Running the aws command rather annoyingly adds control characters. I spent a frustrated hour wondering why the exact text rendered from its output could not be used as input for another command. 

To remove the invisible control characters, run something like:

aws secretsmanager list-secrets --query "SecretList[?Name=='CloudDetails'].ARN" --output text --no-cli-pager | tr -d '[:cntrl:]' | tr -d '\n' 

where in this example, I trying to find the ARN of a particular secret.

Configuring Polaris for Azure

Azure Config

To dispense tokens, you need to register what Azure calls an app. You do this with:

az ad app create \

  --display-name "MyMultiTenantApp" \

  --sign-in-audience "AzureADMultipleOrgs"

The sign-in-audience is an enum that determines if we're allowing single- or multi-tenant access (plus some Microsoft specific accounts).

Create the service principle with:

az ad sp create --id <appId>

where <appId> was spat out by the create command.

Finally, you need to assign roles to the app:

az role assignment create  --assignee <appId>  --role "Storage Blob Data Contributor" --scope /subscriptions/YOUR_SUBSCRIPTION/resourceGroups/YOUR_RESOURCE_GROUP

One last thing: you might need the credentials for this app so you can pass them to Polaris. You can do this with:

az ad app credential reset --id <appId>

and it will tell you the password. This is AZURE_CLIENT_SECRET (see below).

Polaris

First, Polaris needs these environment variables set to access Azure:

AZURE_TENANT_ID

AZURE_CLIENT_ID

AZURE_CLIENT_SECRET

Note that these are the credentials of the user connecting to Azure and not those of the principal that will dispense tokens. That is, they're your app's credentials not your personal ones.

Configuring the Catalog

The values perculiar to Azure that you need to add to the storageConfigInfo that is common to all cloud providers. These are:

• tenantId. You can find this by running az account show --query tenantId.
• multiTenantAppName. This is the Application (client) ID that was generated when the app was created. You can see it in Microsoft Entra ID -> App Registrations -> All Applications in the Azure portal or using the CLI: az ad app list, find your app with the name you created above and use its appId.
• consentUrl. I'm not entirely sure what this is but can be generated with APPID=$(az ad app list --display-name "MyMultiTenantApp" --query "[0].appId" -o tsv) && echo "https://login.microsoftonline.com/common/oauth2/v2.0/authorize?client_id=$APPID&response_type=code&redirect_uri=http://localhost:3000/redirect&response_mode=query&scope=https://graph.microsoft.com/.default&state=12345"

Find out which url to use:

$ az storage account show --name odinconsultants --resource-group my_resource --query "{kind:kind, isHnsEnabled:isHnsEnabled}" -o table

Kind       IsHnsEnabled

---------  --------------

StorageV2  True

HNS stands for hierarchical nested structure.

For StorageV2 and HNS equal to True, use abfss in the allowedLocations part of the JSON sent to /api/management/v1/catalogs.

Testing the credentials

Check the validity of the SAS token with:

az storage blob list  --account-name odinconsultants  --container-name myblob --sas-token $SAS_TOKEN --output table

We get SAS_TOKEN by putting a break point in Iceberg's ADLSOutputStream constructor.

Iceberg bug?

Iceberg asserts that the key in the map of Azure properties sent by Polaris for the expiry time is the string is adls.sas-token-expires-at-ms.PROJECT_NAME. But Polaris is emitting the string expiration-time. I've raised a bug in the Iceberg project here (14069).

I also raised a concurrency bug here (14070) but closed it when I realised it had been fixed in the main branch even if it wasn't in the latest (1.9.2) release.

Anyway, the upshot is that my workaround is to mask the Iceberg code by putting this (almost identical) class first in my classpath.

Configuring Polaris for GCP

After successfully configuring Polaris to vend AWS credentials, I've moved on to the Google Cloud Platform.

As far as the Spark client and Polaris are concerned, there's not much difference to AWS. The only major change is that upon creating the Catalog, you need to put in your JSON a gcsServiceAccount. This refers to the service account you need to create.

You create it with something like:

$ gcloud iam service-accounts create my-vendable-sa --display-name "Vendable Service Account"

and then add to it the user for whom it will deputize:

$ gcloud iam service-accounts add-iam-policy-binding my-vendable-sa@philltest.iam.gserviceaccount.com   --member="user:phillhenry@gmail.com"   --role="roles/iam.serviceAccountTokenCreator"

where philltest is the project and phillhenry is the account for which the token will proxy.

$ gcloud auth application-default login --impersonate-service-account=my-vendable-sa@philltest.iam.gserviceaccount.com

Note it will warn you to run something like gcloud auth application-default set-quota-project philltest or whatever your project is called.

This will open your browser. After you sign in, it will write a credential in JSON file (it will tell you where).  You must point the environment variable GOOGLE_APPLICATION_CREDENTIALS to this file before you run Polaris.

Note that it's the application-default switch that writes the credential as JSON so it can be used by your applications.

This creates a token for you but not that for a service account. For that you need:

gcloud iam service-accounts keys create ~/service-account.json --iam-account my-vendable-sa@philltest.iam.gserviceaccount.com

You might need your friendly admin to run:

gcloud org-policies set-policy allow-key-creation.yaml 

where that YAML file contains:

name: projects/fjords-450014/policies/iam.disableServiceAccountKeyCreation

spec:

  rules:

    - enforce: false

Not so fast...

Running my Spark code just barfed when it tried to write to GCS with:

my-vendable-sa@philltest.iam.gserviceaccount.com does not have serviceusage.services.use access

Note that if the error message mentions your user (in my case PhillHenry@gmail.com) not the service account, you've pointed GOOGLE_APPLICATION_CREDENTIALS at the wrong application_default_credentials.json file.

Basically, I could write to the bucket using the command line but not using the token. I captured the token by putting a breakpoint here in Iceberg's parsing of the REST response from Polaris. Using the token (BAD_ACCESS_TOKEN), I ran:

$ curl -H "Authorization: Bearer ${BAD_ACCESS_TOKEN}" https://storage.googleapis.com/storage/v1/b/odinconsultants_phtest

{

  "error": {

    "code": 401,

    "message": "Invalid Credentials",

    "errors": [

      {

        "message": "Invalid Credentials",

        "domain": "global",

        "reason": "authError",

        "locationType": "header",

        "location": "Authorization"

      }

    ]

  }

}

The mistake I was making was that the command line was associated with me (PhillHenry) not the service account - that's why I could upload on CLI. Check your CLOUDSDK_CONFIG environment variable to see which credential files you're using.

Seems that the service account must have the role to access the account: 

$ gcloud projects get-iam-policy philltest --flatten="bindings[].members"  --format="table(bindings.role)"  --filter="my-vendable-sa@philltest.iam.gserviceaccount.com"

ROLE

roles/serviceusage.serviceUsageAdmin

roles/storage.admin

roles/storage.objectAdmin

roles/storage.objectCreator

This appears to fix it because the service account must be able to see the project before it can see the project's buckets [SO].

Now, Spark is happily writing to GCS.

Iceberg Distributions

There are natural but opposing forces at work in writing data with Iceberg/Spark. Writing many files is more efficient at processing time but leads to greater admin costs - in the extreme, they can cause OOMEs when handling the table's metadata.

The user can fiddle with a few knobs to mitigate this. One is write.distribution-mode. Here are some tests I created to see how the configuration changes affect the number of files when writing the same data:

 

write.distribution-mode	Number of files	Notes
"hash"	p	df.writeTo(tableName).append()
"hash", sorted DataFrame	p	...TBLPROPERTIES ('sort-order' = 'partitionField ASC NULLS FIRST'...
"hash", sorted table	p	df.sort("partitionField").writeTo(tableName).append()
"hash", sorted table but only one value for partitionField	1	because p=1; assumes the size of the data to write is < write.spark.advisory-partition-size-bytes. Otherwise multiple files are written (Spark 3.5).
"none"	d * p	df.writeTo(tableName).append()
"none", sorted DataFrame	p	df.sort("partitionField").writeTo(tableName).append()
"none", sorted table	d * p	...TBLPROPERTIES ('sort-order' = 'partitionField ASC NULLS FIRST'...
"none", sorted table but only one value for partitionField	d	because p=1

p = number of (logical) partitions in the data

d = number of (physical) partitions in the data frame

Note that this is for Spark 3.5. For a distribution mode of hash, and with the size of data exceeding advisory-partition-size-bytes, multiple threads write multiple files.

But for Spark 3.3, if we use a distribution mode of hash and the data exceeds the size of write.spark.advisory-partition-size-bytes, then only one thread writes.

Fan out made no difference in my tests that measured the number of files but it should be used, despite what the documentation says. Contrary to the documentation, Russell Spitzer on Discord says:

"Fanout writer is better in all cases.  We were silly.  The memory requirements were tiny IMHO.  Without fanout, you need to presort within the task but that ends up being way more expensive (and memory intesive) IMHO.  In the latest versions @Anton Okolnychyi removed the local sort requirements if fanout is enabled, so I would recommend fanout always be enabled and especially if you are using distribution mode is none." 

Iceberg Tidbits

Some miscellaneous Iceberg notes.

Cleaning up

There is a logic to this. Russell Spitzer, of the Apache team says:

run rewrite first

expire second

you don't need to run remove_orphans unless something broke [Discord]

Here is a quick BDD I wrote that illustrates what's going on. Basically: 

1. rewrite_data_files puts all the data in as few a number of files as possible. 
2. expire_snapshots then deletes any files that are surplus.

Fewer files means less IO and a more efficient query.

Russell explains the reason we don't use remove_orphans  

To clarify, Remove Orphan Files is only for cleaning up files from failed writes or compactions that failed in a way that the process could not clean up.

Inside of Remove orphan files are 2 expensive remote steps

• Listing everything in the directories owned by the table
• Join of that list with metadata file paths

Then the list of files to delete is brought back to the driver where deletes are preformed [Discord]

This BDD demonstrates removing orphans through the Java API. I wanted to use CALL system.remove_orphan_files but couldn't. Instead, I got the error:

java.lang.IllegalArgumentException: Cannot remove orphan files with an interval less than 24 hours. Executing this procedure with a short interval may corrupt the table if other operations are happening at the same time. If you are absolutely confident that no concurrent operations will be affected by removing orphan files with such a short interval, you can use the Action API to remove orphan files with an arbitrary interval.

at org.apache.iceberg.spark.procedures.RemoveOrphanFilesProcedure.validateInterval(RemoveOrphanFilesProcedure.java:209)

which is a bit much for a simple BDD.

I forced Iceberg to break by having two threads try to write to the same table at the same time.

Pushdown

This is more Spark than just Iceberg but it's an interesting use case.

The IN clause is not pushed down, at least in Spark 3.5.1 - see some nice analysis here. TL;DR; instead of using IN, the most efficient query just converts it into a set of ANDs and ORs. Note that pushdown is (according to Russel Spitzer) pushing the actual function to the data and not mapping it to some other function that is semantically equivalent (as we see in that link).

This filter was not pushed down

Again, I've got a BDD to demonstrate this. Note that if the filters element is empty in a Physical Plan's BatchScan, your function has not been pushed down.

Notes on GPUs and Spark

There are moves afoot to give Spark GPU access out of the box. From Project Hydrogen for Spark: "Although Spark supports [Kubernetes and YARN], Spark itself is not aware of GPUs exposed by them and hence Spark cannot properly request GPUs and schedule them for users. This leaves a critical gap to unify big data and AI workloads and make life simpler for end users."

To play with it, I tried to build Spark Rapids from NVidia. Unfortunately, the tests barfed with:

GpuDeviceManagerSuite:

*** RUN ABORTED ***

  java.lang.OutOfMemoryError: Could not allocate native memory: std::bad_alloc: out_of_memory: RMM failure at:/home/jenkins/agent/workspace/jenkins-spark-rapids-jni-release-10-cuda11/target/libcudf/cmake-build/_deps/rmm-src/include/rmm/mr/device/arena_memory_resource.hpp:181: Maximum pool size exceeded

  at ai.rapids.cudf.Rmm.allocInternal(Native Method)

  at ai.rapids.cudf.Rmm.alloc(Rmm.java:519)

  at ai.rapids.cudf.DeviceMemoryBuffer.allocate(DeviceMemoryBuffer.java:147)

  at ai.rapids.cudf.DeviceMemoryBuffer.allocate(DeviceMemoryBuffer.java:137)

  at com.nvidia.spark.rapids.GpuDeviceManagerSuite.$anonfun$new$4(GpuDeviceManagerSuite.scala:57)

  at com.nvidia.spark.rapids.GpuDeviceManagerSuite.$anonfun$new$4$adapted(GpuDeviceManagerSuite.scala:52)

  at com.nvidia.spark.rapids.TestUtils$.withGpuSparkSession(TestUtils.scala:139)

  at com.nvidia.spark.rapids.GpuDeviceManagerSuite.$anonfun$new$3(GpuDeviceManagerSuite.scala:52)

  at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)

  at org.scalatest.OutcomeOf.outcomeOf(OutcomeOf.scala:85)

OK, so let's use some NVidia tools to find out what's going on. Running nvidia-smi yields:

| NVIDIA-SMI 470.256.02   Driver Version: 470.256.02   CUDA Version: 11.4     |

So running the NVidia profiler, nsight-sys, means I can only profile the CPU not the GPU since the driver is too old (I'm using Ubuntu 20).

It seems [SO] that the CUDA toolkit v12 has a minimum CC of 5 and my Quadro T2000 has a capability factor of 7.5 [NVIDIA] so I should be good if I had an upgraded driver. (Seems that v535 of the NVIDIA driver may have some issues, though).

But it was about this time I bought a new laptop (Lenovo Thinkpad P1 Gen 7 - awesome machine) which came with Ubuntu 22 preinstalled and most of the software I needed. However, nsight-sys was barfing.

The actual error was: Cannot mix incompatible Qt library (5.15.3) with this library (5.15.2)

So, I reinstalled it from the NVIDIA website and now run the command line:

/usr/local/NVIDIA-Nsight-Compute-2024.3/ncu-ui

to see the NSight Compute GUI. There seem to be a few people on the forums suggesting that installing the NVidia tools by hand lead to them being more reliable.

Anyway, I can now run this little handy script:

$ cat ~/bin/nvidia_prof 

/usr/local/NVIDIA-Nsight-Compute-2024.3/target/linux-desktop-glibc_2_11_3-x64/ncu --verbose --config-file off --export /tmp/nvidia.log --force-overwrite  $@

appending it with the code I want to run and I can view the output in ncu-ui and (hopefully) see where the problem is.

Another Java GPU library

It's worth noting that another library in Java that accesses the GPU but this time in a different way. DJL appears to use JNA to access the CUDA library. The interface is CudaLibrary that native implementation of which appears to be a thin wrapper around some Cuda code.

JNA eliminates the boilerplate of JNI. It dynamically maps Java method calls to native functions at runtime. Consequently, JNA has higher runtime overhead compared to JNI because it uses reflection and runtime mapping.

NVIDIA Rapids

You can make Spark go faster by offloading some of the work to the GPU. There is an NVIDIA library (spark-rapids) to do this. 

A quick introduction

There are quite a few references to UCX in Spark Rapids. Assuming you have the hardware, this allows remote direct memory access (RDMA), basically sharing data in memory that circumvents the kernel.

"UVM or universal memory can allow main host memory to act essentially as swap for device(GPU) memory. This allows the GPU to process more data than fits in memory, but  can result in slower processing. This is an experimental feature." (from RapidsConf)

The spillStorageSize is the "Amount of off-heap host memory to use for buffering spilled GPU data before spilling  to local disk. Use -1 to set the amount to the combined size of pinned and pageable memory pools."

Old Ubuntu

Unfortunately, when running a test on Ubuntu 18 (I know, I know) I saw:

Caused by: java.lang.UnsatisfiedLinkError: /home/henryp/Code/Scala/SparkEcosystem/spark-rapids/integration_tests/target/tmp/cudf3561040550923512030.so: /lib/x86_64-linux-gnu/libc.so.6: version `GLIBC_2.28' not found (required by /home/henryp/Code/Scala/SparkEcosystem/spark-rapids/integration_tests/target/tmp/cudf3561040550923512030.so)
...
        at ai.rapids.cudf.NativeDepsLoader.loadDep(NativeDepsLoader.java:246)

After installing the latest Ubuntu on VMWare, it appears that you cannot access the GPU using VMWare Workstation.

You can however use a Docker image - just check that you have the Docker daemon that can handle NVIDIA installed by running:

henryp@adele: docker info | grep Runtimes 

Runtimes: io.containerd.runc.v2 nvidia runc

or 

henryp@adele:~$ grep -A2 -i nvidia /etc/docker/daemon.json

        "nvidia": {

            "path": "/usr/bin/nvidia-container-runtime",

            "runtimeArgs": []

        }

and download an image from here.

Now, I just need to run something like:

docker run --rm --gpus all  --runtime=nvidia  -it   -v /home/henryp/Code/Scala/SparkEcosystem/spark-rapids/:/home/henryp/Code/Scala/SparkEcosystem/spark-rapids/ -v /home/henryp/Tools:/home/henryp/Tools -v /home/henryp/.m2:/.m2  -v /usr/local/bin/Java:/usr/local/bin/Java  --user root  nvidia/cuda:12.6.1-devel-ubi8 /bin/bash

and once I'm in, run:

yum -y install git

yum -y install diffutils

yum -y install rsync

export MAVEN_HOME=/home/henryp/Tools/Build/Maven/Latest

export PATH=$MAVEN_HOME/bin:$PATH

export JAVA_HOME=/usr/local/bin/Java/Latest17

export PATH=$JAVA_HOME/bin:$PATH

Constantly installing some tools for Rapids to build proved a bit tedious, so I extended the NVIDIA docker image with this Dockerfile:

FROM nvidia/cuda:12.6.1-devel-ubi8

RUN yum -y install git

RUN yum -y install diffutils

RUN yum -y install rsync

Also, if I wanted to use mvnDebug [SO], I had to set the network of the container to host using --network host [SO]. Then, it's just a matter of running:

mvnDebug scalatest:test -DforkMode=never

and attaching a debugger from the host machine.

Unfortunately, sometimes when I close and re-open my laptop, the tests start failing with:

...

WindowedBlockIteratorSuite:

terminate called after throwing an instance of 'cudf::jni::jni_exception'

  what():  CUDA ERROR: code 999

/home/henryp/Tools/Build/Maven/Latest/bin/mvnDebug: line 36:   216 Aborted                 (core dumped) env MAVEN_OPTS="$MAVEN_OPTS" MAVEN_DEBUG_OPTS="$MAVEN_DEBUG_OPTS" "`dirname "$0"`/mvn" "$@"

Apparently, I must reboot my machine :(

Some miscellaneous code notes

Finally, I get to look at the code in action. Rapids adds RapidsExecutorPlugin (which extends the Spark ExecutorPlugin interface) that immediately initializes the GPU and memory (see GpuDeviceManager.initializeGpuAndMemory). Note that in setGpuDeviceAndAcquire we see a comment that says: 

"cudaFree(0) to actually allocate the set device - no process exclusive required since we are relying on Spark to schedule it properly and not give it to multiple executors"

This is why the tests (HashAggregatesSuite) a hard-coded to use just one CPU core.

Rapids then has a parallel set of classes that look a lot like the Spark classes that represent linear algebra structures. For example, there is a ColumnVector abstraction in both Spark and Rapids. The interesting Rapids one is GpuColumVector  - which implements this Spark interface - that can be instantiated by a GpuShuffleExchangeExecBase. Amongst other things, objects of these classes contain the address of their off-heap data and a reference counter.

Still playing.

Running Iceberg Catalogs in a test suite

When trying to put some BDDs together here (GitHub) for Iceberg and Spark integration, I hit some snags. I was using the local catalog. If I used the Hive catalog, I was getting this error ("Iceberg does not work with Spark's default hive metastore"). 

The reason is that Derby, which is bundled with the Hive metastore in Spark, "doesn't support the interfaces (or concurrent access) required for Iceberg to work" according to Russell Spitzer. He suggests using the Hadoop catalog instead. 

I didn't fancy setting up a more robust Hive metastore that uses a more "professional" database like Postgres so I  followed Russel's advice to use the Hadoop catalog... and immediately hit this problem that's discussed more in Slack. Basically, CALLing via Spark SQL some procedures throws an IllegalArgumentException stating "Cannot use non-v1 table". This is a Spark v1 table, and nothing to do with Iceberg v1 tables, it seems.

OK, so this got me thinking that if an easy way to use Hive and Hadoop catalogs don't work, let's try a REST catalog. Apache Polaris seemed a good candidate. At first, I was hoping to have it run in-process but because of transitive dependencies, it wasn't possible to have everything running easily in the same JVM.

Hmm, so we need it in a separate process. OK, we can Docker-ise that. After adjusting the fixtures, I noticed that while running an erstwhile passing test for a second time leads to a failure. Looking at the stack traces, it appears that sure, we DROP TABLE IF EXISTS ..., but the underlying files are left behind. Russell Spitzer helped me out (again) by pointing out that a PURGE keyword needs to be added to the SQL.

Does the catalog also need to delete metadata files even if the query engine deletes the data files?

Russell Spitzer

This isn't actually defined anywhere. So at the moment we are kind of in a wild west, the Spark implementation deletes absolutely everything. I believe for Trino it just sends the request to the catalog and it's up to the catalog to decide. So for Spark "no", and it explicitly sends a drop table request to the rest catalog without a purge flag.

Russell says Polaris will delete files with this flag [caveat] although there is debate about making this service asynchronous ("I think polaris could own this service too but I know some folks (like netflix) have delete services already"), and "the option to hook into an external delete service".

So, there is some latitude in how one implements a catalog, it seems. This would explain why some of my tests are failing - some assumptions I made are not guaranteed. Still working on making them pass.

Hive for an Iceberg test suite

I'm currently having trouble changing my catalog implementation from hadoop to hive for my test suite. The tests work fine as they are but they're backed by a local filesystem which is not atomic. For the purposes of testing, this seemed OK but the fact they were not really realistic was nagging me.

Spark can use numerous catalogs. You just need to set them up in SparkConf by giving the config their own namespaces below spark.sql.catalog. Your call to writeTo will use the default namespace unless another is defined.

If you use a Hive catalog, concurrent creation of a table will blow up with:

org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException: [TABLE_OR_VIEW_ALREADY_EXISTS] Cannot create table or view `database`.`concurrentwritespec` because it already exists.

Choose a different name, drop or replace the existing object, or add the IF NOT EXISTS clause to tolerate pre-existing objects.

at org.apache.spark.sql.hive.HiveExternalCatalog.$anonfun$createTable$1(HiveExternalCatalog.scala:255)

Whilte backed by a local file system, the test correctly encounters an error but this time it's:

org.apache.spark.util.Utils - Aborting task

org.apache.iceberg.exceptions.CommitFailedException: Version 1 already exists: /tmp/SparkForTesting5181526160591714739/ConcurrentWriteSpec/metadata/v1.metadata.json

The behaviour for both is entirely expected but the actual Exceptions diff.

Meanwhile, a test that merely appended data blew up (see this test) using hive rather than hadoop failing with:

Cannot write into v1 table: `spark_catalog`.`database`.`spark_file_test_writeto`.

The reason for this currently eludes me but it seems I am not the only one.

Spark Datasource API V1 and V2

Amongst the problem with V1 was "Partitioning and Sorting is not propagated from the data sources, and thus, not used in the Spark optimizer" [Qorela]. V2 gives much greater freedom to developers of external stores, like various cloud solutions. For example Russel Spitzer says:

"Now V2 Predicate function pushdown is allowed (or at least should be coming soon) but for that you must use the datasource functions and not the spark wones" [Slack]

"Best practice for partitions is partition by low cardinality and sort for high cardanality. This may reduce the level of partitions you need" [Slack]

Another key difference between the two is atomicity. "Spark ships with two default Hadoop commit algorithms — version 1, which moves staged task output files to their final locations at the end of the job, and version 2, which moves files as individual job tasks complete...  the v2 Hadoop commit protocol is almost five times faster than v1. This is why in the latest Hadoop release, v2 is the default commit protocol... while v2 is faster, it also leaves behind partial results on job failures, breaking transactionality requirements. In practice, this means that with chained ETL jobs, a job failure — even if retried successfully — could duplicate some of the input data for downstream jobs." [Databricks]

Remediation attempts

Changing the table made no difference to the error:

      spark.sql(

        s"""

          |ALTER TABLE $fqn

          |SET TBLPROPERTIES ('format-version'='2')

      """.stripMargin)

So, I tried creating the table with:

... USING ICEBERG TBLPROPERTIES ('format-version'='2')

only for it to then complain upon table creation:

ERROR | o.a.h.h.metastore.RetryingHMSHandler - MetaException(message:Unable to update transaction database java.sql.SQLSyntaxErrorException: Table/View 'NEXT_LOCK_ID' does not exist.

coming from Hive when it calls Derby. "The Derby catalog doesn't support the interfaces (or concurrent access) required for iceberg to work" [GitHub issue]

So, for the time being at least, it seems that my tests will need to be somewhat unrealistic when it comes to atomicity.

When adding more CPUs does not help distressed CPUs

This is an interesting problem on Discourse where the symptoms belie the cause. Here, a very beefy Spark cluster is taking a long time process (admittedly) a large amount of data. However, it's the CPUs that are getting hammered. 

Insanely high CPU usage

The temptation at this point is to add more CPU resources but this won't help much.

When your Spark jobs that are not computationally intensive are using large amounts of CPU, there's an obvious suspect. Let's check time spent in Garbage Collection:

Insanely large GC Times

Shuffle per worker seems modest but look at those GC Times. In a five hours job, nearly two hours is spent just garbage collecting. 

And this is something of a surprise to people new to Spark. Sure, it delivers on its promise to process more data than can fit in memory but if you want it to be performant, you need to give it as much memory as possible.  

Big Data and CPU Caches

I'd previously posted about how Spark's data frame schema is an optimization not an enforcement. If you look at Spark's code, schemas save checking whether something is null. That is all. 

Can this really make so much of a diffence? Surprisingly, omitting a null check can optimize your code by an order of magnitude. 

As ever, the devil is in the detail. A single null check is hardly likely to make a difference to your code. But when you are checking billions of times, you need to take it seriously. 

There is another dimension to this problem. If you're checking the same reference (or a small set of them) then you're probably going to be OK. But if you are null checking large numbers of references, this is where you're going to see performance degradation.

The reason is that a small number of references can live happily in your CPU cache. As this number grows, they're less likely to be cached and your code will force memory to be loaded from RAM into the CPU.

Modern CPUs cache data to avoid hitting RAM. My 2.40GHz Intel Xeon E-2286M has three levels of cache, each bigger (and slower) than the next:

$ sudo dmidecode -t cache  

Cache Information                       

Socket Designation: L1 Cache                     

Maximum Size: 512 kB                 

...                  

Socket Designation: L2 Cache                   

Maximum Size: 2048 kB                

...                      

Socket Designation: L3 Cache                      

Maximum Size: 16384 kB             

Consequently, the speed we can randomly access an array of 64-bit numbers depends on the size of the array. To demonstrate, here is some code that demonstrates. The results look like this:

Who would have thought little optimizations on big data can make such a huge difference?

Spark and Schemas

I helped somebody on Discord with a tricksy problem. S/he was using a Python UDF in PySpark and seeing NullPointerExceptions. This suggests a Java problem as the Python error message for an NPE looks more like "AttributeError: 'NoneType' object has no attribute ..." But why would Python code cause Spark to throw an NPE?

The problem was the UDF was defining a returnType struct that stated a StructField was not nullable.

The line charge_type.lower (highlighted) was a red herring as they had clearly changed more than one thing when experimenting (always change one thing at a time!)

Note that Spark regards the nullable field as advisory only.

When you define a schema where all columns are declared to not have null values , Spark will not enforce that and will happily let null values into that column. The nullable signal is simply to help Spark SQL optimize for handling that column.

- Spark, The Definitive Guide

And the reason is in this code where Spark is generating bespoke code. If nullable is false, it does not check the reference unnecessarily. But if there reference is null, Spark barfs like so:

Caused by: java.lang.NullPointerException

        at org.apache.spark.sql.catalyst.expressions.codegen.UnsafeWriter.write(UnsafeWriter.java:110)

        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.writeFields_0_2$(Unknown Source)

        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)

        at org.apache.spark.sql.execution.python.EvalPythonExec.$anonfun$doExecute$11(EvalPythonExec.scala:148)

So, the Python returned without an NPE but caused the JVM code to error as the struct it returns contains nulls when it said it wouldn't.

Avoiding Spark OOMEs

Spark can process more data than it can fit into memory. So why does it sometimes fail with OutOfMemoryExceptions when joining unskewed data sets?

An interesting way to counter OOMEs in a large join is here [SO] where rows are given a random integer seed that is used in addition to the usual condition. In theory, this breaks down the data into more manageable chunks.

Another standard exercise is to repartition the data. But this causes a shuffle and it may actually be the repartition itself that causes of an OOME.

In practice, I've found persisting the data frame to disk and reading it back yields better results. The number of partitions being written is rarely the number that is read back. That is, you get a more natural partition for free (or almost free. Obviously, some time is taken in writing to disk). And there is no repartition that could throw an OOME.

This question came up on Discord where somebody is trying to crossJoin a huge amount of data. I suggested a solution that uses mapPartitions. The nice thing about this method is that your code is passed a lazy data structure. As long as you don't try to call something like toList on it, it will pull data into memory as needed and garbage collect it after it's written out.

By using a lazy Iterator, Spark can write far more memory than it has to disk. As Spark consumes from the Iterator, it measures its memory. When it starts looking a bit full, it flushes to disk. Here is the memory usage of this code that uses mapPartitions to write to /tmp/results_parquet a data set that is much larger than the JVMs heap:

Spark with 0.5gb heap writing 1.3gb files

If we run:

watch "du -sh /tmp/results_parquet"

we can see that upon each GC, more is written to disk.

The result is a huge dataframe that could not fit into memory can now be joined with another.

As an aside: Uber has been doing some work on dealing with OOMEs in Spark. See their article here. TL;DR; they're proposing that in the event of an OOME, Spark adapts and increases the memory to CPU ratio by asking come cores to step down before it re-attempts the failed stage. Ergo, each compute unit has more memory than before. 

Z-order

Z-ordering is an optimization technique in big data that allows faster access since similar data lives together. We discuss the algorithm that defines what is similar here. 

Imagine a logical grid where all the values of one column run across the top and all the values from another run down the side. If we were to sort this data, every datum can be placed somewhere in that grid.

Now, if the squares of the grid were mapped to files and all the data in each cell were to live in those files, we have made searching much easier as we now know the subset of files in which it may live. We've essentially sorted in not just one dimension but two (although we can do higher).

This can be especially useful when we want to sort the data but don't know exactly what to sort on - a common connundrum when dealing with events. Say we have event_time, insert_time and update_time. Which do we choose? We could sort the data three times, each time on one column but this is impractical with huge data sets. Enter z-order.

Note that the Z-Order really needs 2 or more columns on which to act. Only one column is the degenerate case. "Zorder/Hilbert etc on a single dimension are just a hierarchal sort" [Russell Spitzer on Slack].

(This is a good article about z-ordering from the perspective of Apache Iceberg.)

For an example in Delta Lake, we can see this code that creates a data set with columns c1, c2 and c3 whose values are [x, 99-x, x+50 mod 100] for x [0, 99]. After z-ordering it, these numbers are split into 4 different files. Generating a graphic illustrates how the data points are distributed over those files:

The idea behind how we calculate which cell a datum falls into is best described here on Wikipedia. But, in brief, the binary representation of the data points is interleaved to give a z-value per tuple. In our example, I see a [0, 99, 50] mapped to the byte array [0, 0, 0, 0, 0, 0, 0, 0, 0, 9, -112, 26].

I took a look at the Delta Lake code here where a Spark Column object is created that wraps a DL InterleaveBits type which in turn is a subclass of Spark's Catalyst Expression type. This executes on Spark's InternalRow, that is, the raw data on the executors.

The reason the code is doing this is to add a column with which we can repartition the data with the SQL CAST(interleavebits(rangepartitionid(c1), rangepartitionid(c2), rangepartitionid(c3)) AS STRING). The rangepartitionid keyword is part of the Delta Lake machinery.

Using this z-order value (plus a random key), the DataFrame the Delta code now calls repartitionByRange which samples the data [SO] and breaks it into discrete ranges.

Given the interleaving of the columns c1, c2 and c3 their order has minimal impact on the z-value so it's no surprise to see nearby data clustering into the same files, as we can see in the graphic. In fact, if you look at the DataFrame during the repartition process:

+---+---+---+-------------------------------------------+
| c1| c2| c3|c7b6b480-c678-4686-aa99-283988606159-rpKey1|
+---+---+---+-------------------------------------------+
|  0| 99| 50|                                       \t�|
|  1| 98| 51|                                       \t�|
|  2| 97| 52|                                       \t�b|
|  3| 96| 53|                                       \t�e|
|  4| 95| 54|                                       \b��|
|  5| 94| 55|                                       \b��|
|  6| 93| 56|                                       \b��|
|  7| 92| 57|                                       \b��|
|  8| 91| 58|                                       \b�|
|  9| 90| 59|                                       \b�|
| 10| 89| 60|                                       \b�b|
| 11| 88| 61|                                       \b�e|
| 12| 87| 62|                                       \b��|
| 13| 86| 63|                                       \b��|
| 14| 85| 64|                                       \f)�|
| 15| 84| 65|                                       \f)�|
| 16| 83| 66|                                       \f`|
| 17| 82| 67|                                       \f`|
| 18| 81| 68|                                       \f`b|
| 19| 80| 69|                                       \f`e|
+---+---+---+-------------------------------------------+

you can see the slowly changing values by which things are partitioned (column c7b6b480-c678-4686-aa99-283988606159-rpKey1 -  a random name so it doesn't clash with other column names. It's dropped immediately after the call to repartitionByRange)

Dependency hell

In these days of ChatGPT, it's easy to forget that most of the time, a developer isn't actually cutting code at all, but debugging it. This is my own personal hell in getting Spark and Kafka in Docker containers talking to a driver on the host.

Firstly, I was seeing No TypeTag available when my code was trying to use the Spark Encoders. This SO answer helped. Basically, my code is Scala 3 and "Encoders.product[classa] is a Scala 2 thing. This method accepts an implicit TypeTag. There are no TypeTags in Scala 3". Yikes. This is probably one reason the upgrade path in Spark to Scala 3 is proving difficult. The solution I used was to create a SBT sub Project that was entirely Scala 2 and from here I called Spark.

The next problem was seeing my Spark jobs fail with:

Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 6) (172.30.0.7 executor 0): java.lang.ClassCastException: cannot assign instance of scala.collection.generic.DefaultSerializationProxy to field org.apache.spark.sql.execution.datasources.v2.DataSourceRDDPartition.inputPartitions of type scala.collection.immutable.Seq in instance of org.apache.spark.sql.execution.datasources.v2.DataSourceRDDPartition

This is a contender for the error message with the greatest misdirection. You think it's a serialization problem but it isn't directly so.

Although other Spark users have reported it, Ryan Blue mentions that it isn't really a Spark issue but a Scala issue.

Anyway, I tried all sorts of things like change my JDK (note the sun.* packages have been removed in later JDKs so you need to follow the advice in this SO answer). I tried creating an uber jar but was thwarted by duped dependencies [SO], Invalid signature file digest errors as some were signed [SO] that forced me to strip the signtures out [SO] but still falling foul of Kafka's DataSourceRegister file being stripped out [SO].

The first step in the right direction came from here and another SO question where the SparkSession is recommended to be built by adding .config("spark.jars", PATHS) where PATHS is a comma delimited string of the full paths of all the JARs you want to use. Surprisingly, this turned out to include Spark JARs themselves, including in my case spark-sql-kafka-0-10_2.13 which oddly does not come as part of the Spark installation. By adding them as spark.jars, they are uploaded into the work subdirectory of a Spark node.

After this, there was just some minor domain name mapping issues to clear up in both the host and container before the whole stack worked without any further errors being puked.

Spark, Kafka and Docker

I want to run a Spark Structured Streaming application that consumes from a Kafka cluster all within Docker. I've finally got it working [messy code here in my GitHub], but it was not without its own pain.

The biggest problem is getting all the components talking to each other. First, you need a bridge network. "In terms of Docker, a bridge network uses a software bridge which allows containers connected to the same bridge network to communicate, while providing isolation from containers which are not connected to that bridge network." [docs]. Think of it as giving your containers their own namespace.

Secondly, the Spark worker needs to connect to Kafka, the Spark master and the Spark driver. The first two are just a matter of mapping the Spark master and Kafka containers in the worker. What's harder is getting the worker to talk to the driver that may be running on the computer that hosts Docker.

One sign you've got it wrong is if you see "Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources" [SO] in your driver logs. This message is a little ambiguous as it may have nothing to do with resources but connectivity. 

Resources seem to be fine

To solve it, you need the driver to define spark.driver.host and spark.driver.port. For the host, we need it to be the magic address of 172.17.0.1. This is the default "IP address of the gateway between the Docker host and the bridge network" [docs]. The port is arbitrary.

[Aside: it's also worth ensuring that the all components are running the exact same version of Spark. I saw a rare error ERROR Inbox: Ignoring error java.lang.AssertionError: assertion failed: CPUs per task should be > 0 and the only thing Google produced was this Bitnami ticket. Ensuring all version were the same made it go away.]

What's more, the worker needs these in its config. You can pass it the host and port with something like SPARK_WORKER_OPTS="-Dspark.driver.host=172.17.0.1 -Dspark.driver.port=SPARK_DRIVER_PORT" in its start up script.

But there is one last gotcha. If still can't get things to work, you might want to login to your worker container and run netstat. If you see the connection to the driver in a state of SYN_SENT, your firewall on the host is probably blocking the connection from the container.

Annoyingly, you probably won't see any error messages being puked from the Driver. It will just hang somwhere near org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:929). I only started seeing error messages when I aligned all version of Spark (see above) and it read: java.io.IOException: Connecting to /172.17.0.1:36909 timed out (120000 ms) 

Looking in that Docker container showed:

bash-5.0# netstat -nap
Active Internet connections (servers and established)
Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name   
...
tcp        0      1 192.168.192.7:53040     172.17.0.1:36909        SYN_SENT    1109/java

and on the host machine where my process is 29006:

(base) henryp@adele:~$ netstat -nap | grep 36909

tcp6       0      0 172.17.0.1:36909        :::*                    LISTEN      29006/java  

Aha, that looks like the problem. It turns out that I have to open the firewall for the block manager too and set a static port for it on the Driver with spark.driver.blockManager.port.

Finally, you should be able to have a Spark master and worker plus Kafka instances all running within Docker along with the driver running on the host using your favourite IDE.