Configuring Polaris Part 1

To vend credentials, Polaris needs an AWS (or other cloud provider) account. But what if you want to talk to several AWS accounts? Well this ticket suggests an interesting workaround. It's saying "yeah, use just one AWS account but if you need to use others, set up a role that allows access to other AWS accounts, accounts outside the one that role lives in."

We are working in a cross cloud environment. We talk to not just AWS but GCP and Azure clouds. We happen to host Polaris in AWS but this choice was arbitrary. We can give Polaris the ability to vend credentials for all clouds no matter where it sits.

Integration with Spark

It's the spark.sql.catalog.YOUR_CATALOG.warehouse SparkConf value that identifies the Polaris catalog.

The YOUR_CATALOG defines the namespace. In fact, the top level value, spark.sql.catalog.YOUR_CATALOG, tells Spark which catalog to use (Hive, Polaris, etc).

So, basically, your config should look something like:

spark.sql.catalog.azure.oauth2.token                                            POLARIS_ACCESS_TOKEN

spark.sql.catalog.azure.client_secret                                                         s3cr3t

spark.sql.catalog.azure.uri                                        http://localhost:8181/api/catalog

spark.sql.catalog.azure.token                                                  POLARIS_ACCESS_TOKEN

spark.sql.catalog.azure.type                                                                    rest

spark.sql.catalog.azure.scope                                                     PRINCIPAL_ROLE:ALL

spark.sql.catalog.azure.client_id                                                               root

spark.sql.catalog.azure.warehouse                                                              azure

spark.sql.catalog.azure.header.X-Iceberg-Access-Delegation                        vended-credentials

spark.sql.catalog.azure.credential                                                       root:s3cr3t

spark.sql.catalog.azure.cache-enabled                                                          false

spark.sql.catalog.azure.rest.auth.oauth2.scope                                    PRINCIPAL_ROLE:ALL

spark.sql.catalog.azure                                        org.apache.iceberg.spark.SparkCatalog 

This is the config specific to my Azure catalog. AWS and GCP would have very similar config.

One small issue [GitHub] is that I needed the Iceberg runtime lib to be the first in the Maven dependencies.  

Local Debugging

Put:

"-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005",

in build.gradle.kts in the 

tasks.named<QuarkusRun>("quarkusRun") {

  jvmArgs =

    listOf(

section then run with:

./gradlew --stop && ./gradlew run

then you'll then be able to remotely debug by attaching to port 5005.

Configuring Polaris in a remote environment

Note that Polaris heavily uses Quarkus. "Quarkus aggregates configuration properties from multiple sources, applying them in a specific order of precedence." [docs]. First, java -D... properties, environment variables, application.properties (first on the local filepath then in the dependencies) and finally the hard-coded values.

Polaris in integration tests

Naturally, you're going to want to write a suite of regression tests. This is where the wonderful TestContainers shines. You can fire up a Docker container of Polaris in Java code.

There are some configuration issues. AWS and Azure are easy to configure within Polaris. You must just pass them the credentials as environment variables. GCP is a little harder as it's expecting a file of JSON containing its credentials (the Application Default Credentials file). Fortunately, TestContainers allows you to copy that file over once the container has started running.

          myContainer = new GenericContainer<>("apache/polaris:1.1.0-incubating")

                    // AWS

                    .withEnv("AWS_ACCESS_KEY_ID",     AWS_ACCESS_KEY_ID)

                    .withEnv("AWS_SECRET_ACCESS_KEY", AWS_SECRET_ACCESS_KEY)

                    // Azure

                    .withEnv("AZURE_CLIENT_SECRET", AZURE_CLIENT_SECRET)

                    .withEnv("AZURE_CLIENT_ID",     AZURE_CLIENT_ID)

                    .withEnv("AZURE_TENANT_ID",     AZURE_TENANT_ID)

                    // Polaris

                    .withEnv("POLARIS_ID",     POLARIS_ID)

                    .withEnv("POLARIS_SECRET", POLARIS_SECRET)

                    .withEnv("POLARIS_BOOTSTRAP_CREDENTIALS", format("POLARIS,%s,%s", POLARIS_ID, POLARIS_SECRET))

                    // GCP

                    .withEnv("GOOGLE_APPLICATION_CREDENTIALS", GOOGLE_FILE)

                    .waitingFor(Wait.forHttp("/q/health").forPort(8182).forStatusCode(200));

            ;

            myContainer.setPortBindings(List.of("8181:8181", "8182:8182"));

            myContainer.start();

            myContainer.copyFileToContainer(Transferable.of(googleCreds.getBytes()), GOOGLE_FILE);

The other thing you want for a reliable suite of tests is to wait until Polaris starts. Fortunately, Polaris is cloud native and offers a health endpoint which TestContainers can poll.

Polaris in EKS

I found I had to mix both AWS's own library (software.amazon.awssdk:eks:2.34.6) with the official Kubernetes library (io.kubernetes:client-java:24.0.0) before I could interrogate the Kubernetes cluster in AWS from my laptop and look at the logs of the Polaris container. 

        EksClient eksClient = EksClient.builder()

                                       .region(REGION)

                                       .credentialsProvider(DefaultCredentialsProvider.create())

                                       .build();

        DescribeClusterResponse clusterInfo = eksClient.describeCluster(

                DescribeClusterRequest.builder().name(clusterName).build());

        AWSCredentials awsCredentials = new BasicAWSCredentials(

                AWS_ACCESS_KEY_ID,

                AWS_SECRET_ACCESS_KEY);

        var authentication = new EKSAuthentication(new STSSessionCredentialsProvider(awsCredentials),

                                                   region.toString(),

                                                   clusterName);

        ApiClient client = new ClientBuilder()

                .setBasePath(clusterInfo.cluster().endpoint())

                .setAuthentication(authentication)

                .setVerifyingSsl(true)

                .setCertificateAuthority(Base64.getDecoder().decode(clusterInfo.cluster().certificateAuthority().data()))

                .build();

        Configuration.setDefaultApiClient(client);

Now you'll be able to query and monitor Polaris from outside AWS's Kubernetes offering, EKS.

Three things about Docker

Most of the time, Docker just works. But sometimes, you need to be a bit clever. In my case, I want Polaris to have he permission to write to the host filesystem, not just to see it. This proved hard. These are some lessons I learned.

What's in an image?

You can break down what is in a Docker image with something like:

docker save ph1ll1phenry/polaris_for_bdd:latest -o polaris_for_bdd.tar

mkdir polaris_fs

tar -xf polaris_for_bdd.tar -C polaris_fs

Then you can start untaring the blobs (where each blob is a Docker layer). In my case, I was trying to find where the bash binary was:

for BLOB in $(ls  blobs/sha256/ ) ; do { echo $BLOB ; tar -vxf blobs/sha256/$BLOB | grep bash;  } done

How was an image built?

You can reconstitute the steps made to create a Docker image with something like:

docker history --no-trunc apache/polaris

Restoring OS properties

The apache/polaris Docker image had a lot of extraneous Linux binaries removed, presumably to make it smaller and more secure. However, I needed them back as I need to grant the container certain permissions on the host. 

First off, the su command had been removed. You can canibalise binaries from other images in your Dockerfile like this:

FROM redhat/ubi9 AS donor

FROM apache/polaris AS final

...

COPY --from=donor /usr/bin/su /usr/bin/su

However, copying a binary over most of the time is a bit naive. Running su gave:

su: Critical error - immediate abort

Taking the parent Docker image before it was pruned, I could run:

[root@6230fb595115 ~]# ldd /usr/bin/su

linux-vdso.so.1 (0x00007fff87ae6000)

libpam.so.0 => /lib64/libpam.so.0 (0x0000733267a15000)

libpam_misc.so.0 => /lib64/libpam_misc.so.0 (0x0000733267a0f000)

libc.so.6 => /lib64/libc.so.6 (0x0000733267807000)

libaudit.so.1 => /lib64/libaudit.so.1 (0x00007332677d3000)

libeconf.so.0 => /lib64/libeconf.so.0 (0x00007332677c8000)

libm.so.6 => /lib64/libm.so.6 (0x00007332676ed000)

/lib64/ld-linux-x86-64.so.2 (0x0000733267a39000)

libcap-ng.so.0 => /lib64/libcap-ng.so.0 (0x00007332676e2000)

So, my Dockerfile had to COPY these files over too.

These libpam* shared objects refer to Linux's Pluggable Authentication Modules which is a centralized framework for permissioning arbitrary modules - eg MFA.

After a lot of faffing, I just COPYd the entire /etc/ folder from the donor to the final images. This is fine for integration tests but probably best avoided for prod :)

Debugging Polaris in Docker

Iceberg and Polaris sitting in a tree...

I have a proof of concept on GitHub that demonstrates how to use Apache Iceberg. Since I need Apache Polaris as Iceberg's metastore, I have it running in a container. 

If I create a catalog of FILE type, if the file path for storage is X, does this refer to the filesystem of Polaris or Spark?

Michael Collado

it's going to be both, so you shouldn't really use it in that scenario. The Polaris server is going to read/write metadata.json files in its own container's file system and the spark notebook will read/write data files in its own container's filesystem, so... [Discord]

In my PoC, I use a shared filesystem mount where both the Polaris container writes as well as the host's Spark instance.

However, tests were failing with the minimum of logging. When running Docker as a non-root user, the error in the Polaris logs looks like:

{"timestamp":1734433498899,"level":"INFO","thread":"dw-51 - POST /api/catalog/v1/manual_spark/namespaces/my_namespace/tables/IcebergCRUDSpec","logger":"org.apache.polaris.service.exception.IcebergExceptionMapper","message":"Handling runtimeException Failed to get file system for path: file:/tmp/polaris/my_namespace/IcebergCRUDSpec/metadata/00000-0daa8a08-5b5d-459a-bdd0-0663534f2007.metadata.json","mdc":{"spanId":"6ea71bffea6af726","traceId":"8b485bf56e7e27aac2e47ede876e02bd","realm":"default-realm","request_id":null},"params":{}}

When running containerised Polaris as root, the tests passed but I couldn't clean up the files on the shared filesystem mount afterwards as I was not running the test suite as root on the hosts.

Digging Deeper

That string ("Failed to get file system for path") lead me to org.apache.iceberg.hadoop.Util.getFs. Unfortuately, the nested exception is wrapped in the error reported above and lost.

So, we start the container with these flags:

 -eJAVA_OPTS=-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:8788 -p8788:8788

since polaris-dropwizard-service is part expecting JAVA_OPTS to be set. 

Great, now we can put a breakpoint in Util.getFs and printStackTrace on the nested exception. It shows:

Caused by: javax.security.auth.login.LoginException: java.lang.NullPointerException: invalid null input: name

        at jdk.security.auth/com.sun.security.auth.UnixPrincipal.<init>(UnixPrincipal.java:71)

        at jdk.security.auth/com.sun.security.auth.module.UnixLoginModule.login(UnixLoginModule.java:134)

        at java.base/javax.security.auth.login.LoginContext.invoke(LoginContext.java:754)

        at java.base/javax.security.auth.login.LoginContext$4.run(LoginContext.java:678)

        at java.base/javax.security.auth.login.LoginContext$4.run(LoginContext.java:676)

        at java.base/java.security.AccessController.doPrivileged(AccessController.java:714)

        at java.base/javax.security.auth.login.LoginContext.invokePriv(LoginContext.java:676)

        at java.base/javax.security.auth.login.LoginContext.login(LoginContext.java:587)

        at org.apache.hadoop.security.UserGroupInformation$HadoopLoginContext.login(UserGroupInformation.java:2148)

A quick look at the JDK code shows that UnixSystem.getUsername appears to be returning a null. And this appears to be because there is no user with my ID in the container - d'oh.

A Solution

One solution is to have a bespoke Docker entrypoint that creates the user (if it doesn't exist) given a user ID passed by the Fabric8's docker-maven-plugin and runs Polaris as that user. If it's the same user as that running the integration tests, both Polaris and Spark can write to the same directory and tables can be dropped without permission issues.

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.

Kafka Raft in Docker

These days, you don't need Zookeeper to run a Kafka cluster. Instead, when correctly configured, Kafka uses the Raft algorithm (where "the nodes trust the elected leader"[Wikipedia]) to coordinate itself.

I started to follow Gunnar Morling's blog but it seems his version of Kafka containers have not been updated so I used Bitnami's. However, configuring them to run a Raft cluster proved difficult.

I want to programatically create the cluster rather than use docker-compose as I want greater control over it. So, I wrote this code that talks to Docker via it's API using a Java library. 

Firstly, the Kafka instances couldn't see each other. 

Diagnosing the containers proved difficult as I could not install my favourite Linux tools. When I tried, I was told directory /var/lib/apt/lists/partial is missing. This seems to be deliberate as the Dockerfile explicitly deletes it, to keep images slim. So, I took out that line and added:

RUN apt-get update && apt-get upgrade -y && \
    apt-get clean && apt-get update && \
    apt-get install -y net-tools && \
    apt-get install -y iputils-ping && \
    apt-get install -y  procps && \
    apt-get install -y lsof

then rebuilt the containers. [Aside: use ps -ax to see all the processes in these containers. I was stumped for a while not seeing the Java process that I knew was running].

Using these Linux tools, I could see the containers could not even ping each other. Oops, I need to create a Docker network [SO] and add it to the containers. Now, their logs show that the Kafka containers are at least starting and talking to each other. 

However, the client running on the host machine was puking lots of messages like "Cancelled in-flight API_VERSIONS request with correlation id 1 due to node -1 being disconnected". First, I checked [SO] that the Kafka client library and container were both version 3.  But the consensus on the internet appears to be that this error is due to a connection failure. 

Using netstat on the host showed that the host port was indeed open. But this seemed to be due to Docker opening the port to map it to its container but the container not LISTENing on that port. It appears you can tell Kafka on which port to listen with an environment variable that looks like:

KAFKA_CFG_LISTENERS=PLAINTEXT://:$hostPort,CONTROLLER://:$controllerPort

where hostPort is what you want Docker to map and controllerPort corresponds to what is in the KAFKA_CFG_CONTROLLER_QUORUM_VOTERS environment variable.

The next problem was when my client connects, it cannot see the machine called kafka2. What's happening here is that having connected to the bootstrap, the client is asked to contact another machine, in this case something called kafka2. 

Now, the JVM running on the host knows nothing about a network that is internal to Docker. To solve this, you could have Docker use the host network (which means that everything running on the machine can see each other - fine for testing but a security nightmare). You could subvert the JVM's DNS mappings (rather than faffing around with a DNS proxy) using BurningWave or Java 18's InetAddressResolverProvider. But perhaps the simplest way is configuring Kafka itself to advertise itself as localhost [Confluent] using the KAFKA_CFG_ADVERTISED_LISTENERS environment variable.

And that was it: a Kafka cluster running on my laptop that was reading and writing messages using the Raft algorithm. There are still a few lose ends: why on some runs a node drops out of the cluster non-deterministically even if the functionality was correct as far as the client was concerned. I'll solve that another day.

Packaging Python

Java programmers don't know the meaning of classpath hell until they've played with Python. Here are some notes I took while ploughing through the excellent Practical MLOps (Gift & Deza). Following their instructions, I as attempting to get a ML model served using Flask in a Docker container. Spoiler: it didn't work out of the box.

Since the correct OnnxRuntime wheel for my Python runtime did not exist, I had to build onnxruntime with --build-wheel while making the artifact.

This is where I encountered my first dependency horror:

CMake 3.18 or higher is required.  You are running version 3.10.2

when running onnxruntime/build.sh. (You can put a new version first in your PATH and avoid having to install it at the OS level).

This finally yielded onnxruntime-1.12.0-cp36-cp36m-linux_x86_64.whl which could be installed into my environment with pip install WHEEL_FILE... except that cp number must correspond to your Python version (3.6 in this case).

Moving virtual environments between machines is hard. You'd be best advised to use pip freeze to capture the environment. But ignoring this advice yields an interesting insight into the Python dependency system:

The first problem is that if you've created the environment with python -m venv then the scripts have your directory structure backed into them, as a simple grep will demonstrate. Copying the entire directory structure up to the virtual environment solved that.

But running the code gave me "No module named ..." errors. Looking at the sys.path didn't show my site-packages [SO] despite me having run activate. Odd. OK, so I defined PYTHONPATH and then I could see my site-packages in sys.path.

Then, you want to use exactly the same Python version. No apt-get Python for us! We have to manually install it [SO]. When doing this on a Docker container, I had to:

RUN apt-get update

RUN apt-get install -y wget

RUN apt-get install -y gcc

RUN apt-get install -y make

RUN apt-get install -y zlib1g-dev

Note that this [SO] helped me to create a Docker container that just pauses the moment it starts. This allows you to login and inspect it without it instantly dying on a misconfiguration.

The next problem: there are many compiled binaries in your virtual environment.

# find $PYTHONPATH/ -name \*.so | wc -l

185

Copying these between architectures is theoretically possible but the "as complexity of the code increases [so does] the likelihood of being linked against a library that is not installed" [SO]

Indeed, when I ran my Python code, I got a Segmentation Fault which can happen if "there's something wrong with your Python installation." [SO]

Python builds

A quick addendum on the how Python builds projects: the standard way is no longer standard: "[A]s of the last few years all direct invocations of setup.py are effectively deprecated in favor of invocations via purpose-built and/or standards-based CLI tools like pip, build and tox" [Paul Gannsle's blog]

Azure, Docker and K8s

I'm trying to get Spark to work on Azure in a Kubernetes container. With Azure, you provision some boxes that have K8s running on them but it's the provisioned boxes you pay for irrespective of the containers running (or not) in Kubernetes.

This gives us greater control over what is deployed and when. However, talking to Azure storage has not been easy.

Talking to Azure Storage with Java drivers

First, I want to use the latest image of Spark. At the time of writing, this is v3.0.0-preview2 but I couldn't find a Docker image for it so I built my own.

$ git checkout v3.0.0-preview2
$ mvn clean install -DskipTests

I then build a Docker image using Spark's ./bin/docker-image-tool.sh and publish it to my Docker hub account. With some K8s config that looks a little like this, I can start up a Spark cluster. In the Spark REPL, I run some fairly hacky code to give me access to Azure:

val accountName = ???
val accountKey  = ???
val container   = ???
val confKey     = s"fs.azure.account.key.$accountName"
val confKey2    = s"$confKey.blob.core.windows.net"
val confKey3    = s"$confKey.dfs.core.windows.net"
val confKey4    = s"$confKey.file.core.windows.net"
val sas         = ???

spark.conf.set( s"fs.azure.sas.$container.$accountName.dfs.core.windows.net", sas)
spark.conf.set( s"fs.azure.sas.$container.$accountName.file.core.windows.net", sas)
sc.hadoopConfiguration.set( s"fs.azure.sas.$container.$accountName.dfs.core.windows.net", sas)
sc.hadoopConfiguration.set( s"fs.azure.sas.$container.$accountName.file.core.windows.net", sas)

spark.conf.set( confKey,  accountKey)
spark.conf.set( confKey2, accountKey)
spark.conf.set( confKey3, accountKey)
spark.conf.set( confKey4, accountKey)

val clazz = "org.apache.hadoop.fs.azure.NativeAzureFileSystem"
sc.hadoopConfiguration.set("fs.abfs.impl",  clazz)
sc.hadoopConfiguration.set("fs.abfss.impl", clazz)
sc.hadoopConfiguration.set("fs.azure",      clazz)
sc.hadoopConfiguration.set("fs.wasbs.impl", clazz)
sc.hadoopConfiguration.set("fs.wasb.impl",  clazz)
sc.hadoopConfiguration.set(confKey,         accountKey)
sc.hadoopConfiguration.set(confKey2,        accountKey)
sc.hadoopConfiguration.set(confKey3,        accountKey)
sc.hadoopConfiguration.set(confKey4,        accountKey)

You can get the credentials you need by running:

$ az storage account keys list -n ACCOUNT_NAME -g GROUP --subscription YOUR_AZURE_SUBSCRIPTION

and the SAS (Shared Access Signature) from the Azure web console (you may need to allow HTTP). Without the SAS, you may get what is probably the most uninformative error message I have ever seen:

Value for one of the query parameters specified in the request URI is invalid.

Which parameter and why is it invalid? This originates on the server side so there is no chance to debug it.

Anyway, when I tried to read from an Azure File System using something like this:

val df = spark.read.parquet(s"abfs://$container@$accountName.dfs.core.windows.net/MY_PARQUET_FILE")

I saw a stack trace puked with:

... Caused by: com.microsoft.azure.storage.StorageException: The specified Rest Version is Unsupported.

Sniffing the network traffic with:

$ tcpdump -A -nn host MY_BOX_IP and MICROSOFTS_API_IP_ADDRESS -i eth0

Was showing:

14:45:32.445119 IP MY_BOX_IP.59274 > MICROSOFTS_API_IP_ADDRESS.80: Flags [P.], seq 1:603, ack 1, win 502, options [nop,nop,TS val 1401020617 ecr 1846141065], length 602: HTTP: HEAD ...
...
Accept: application/xml
Accept-Charset: UTF-8
Content-Type:
x-ms-version: 2014-02-14
User-Agent: Azure-Storage/2.0.0 (JavaJRE 1.8.0_242; Linux 4.15.0-1066-azure)
x-ms-client-request-id: dfd4qall-6657-45f7-9ed5-00e455e95bee
Host: MY_BOX.dfs.core.windows.net
Connection: keep-alive

This is an ancient version. Spark by default depends on Hadoop 2.7.4 which pulls in azure-storage 2.0.0 (see hadoop/hadoop-project/pom.xml), a very old version.

OK, so let's rebuild Spark with this:

mvn clean install -DskipTests -Phadoop-3.2

The hadoop-3.2 profile gives me a dependency on, you guessed it, a later version of Hadoop that provides the transitive dependency of azure-storage 7.0.0 that dates to February 2018.

I push this to my Docker Hub account with:

$ ./bin/docker-image-tool.sh  -r docker.io/ph1ll1phenry -t spark3.0.0-preview2_hadoop3.2.0 build

$ docker images | grep spark3.0.0-preview2_hadoop3.2.0

ph1ll1phenry/spark                         spark3.0.0-preview2_hadoop3.2.0   931173a555b6        About a minute ago   545MB

$ docker tag 931173a555b6 ph1ll1phenry/spark3.0.0-preview2_hadoop3.2.0

$ docker push ph1ll1phenry/spark3.0.0-preview2_hadoop3.2.0

don't forget to tag it (SO) and deploy my cluster on Kubernetes as before.

But this too appears to be an old client as running against Azure Storage results in the same exception despite tcpdump now showing the relevant HTTP header as:

x-ms-version: 2017-07-29

Some crumb of comfort is that I can read the file if I use a slightly different URL.

val df = spark.read.parquet(s"wasb://landing@$accountName.blob.core.windows.net/MY_PARQUET")

However, any writing results in:

com.microsoft.azure.storage.StorageException: This operation is not permitted on a non-empty directory.

even when the directory is not empty at all.

Another soupcon of good news is that at least I can read and write from and to Azure Blob Containers:

val df = spark.read.text(s"wasbs://$container@$accountName.blob.core.windows.net/MY_PARQUET")

This works.

What's the version number, Kenneth?

In desperation, I forced Spark to use a more recent version by running Spark with:

--packages org.apache.hadoop:hadoop-azure:3.2.0,com.microsoft.azure:azure-storage:8.6.0

And sure enough, tcpdump shows:

x-ms-version: 2019-02-02

This seems to pass the x-ms-version check but then results in:

Incorrect Blob type, please use the correct Blob type to access a blob on the server. Expected BLOCK_BLOB, actual UNSPECIFIED.

which (I'm guessing) is due to incompatibility due to azure-storage not being the version hadoop-azure is expecting.

I've left a message on the Hadoop users' mailing list asking for help as I still want to talk to the Azure File System rather than a Blob Container.

Hacky workaround

To get a Linux Azure instance that can mount Gen 2 storage, check out this GitHub repository.

I changed the kubernetes-volume-drivers/flexvolume/blobfuse/deployment/Dockerfile so:

FROM ubuntu:18.04

...

RUN apt update && apt install -y openjdk-8-jdk

giving me a newer OS and installing Java. Then, I deploy the OS image to Docker Hub:

$ cd kubernetes-volume-drivers/flexvolume/blobfuse/deployment

$ docker build  -t blobfuse-jdk8 blobfuse-flexvol-installer/

Now let's build and deploy it: 

$ docker images | grep blobfuse-jdk8

$ docker tag IMAGE_TAG ph1ll1phenry/blobfuse-openjdk-8-jdk-slim

$ docker push ph1ll1phenry/blobfuse-openjdk-8-jdk-slim

and having a look at Docker Hub, I can see my OS image.

Now, we need to get Spark to use this OS, so I slightly bastardise its Dockerfile:

$ git diff
...
-FROM openjdk:8-jdk-slim
+FROM ph1ll1phenry/blobfuse-openjdk-8-jdk-slim

We build it, docker push it and kubernetes apply the a slightly different yaml file and now Spark has a mount onto the Azure File System at /mnt/data.

Note that you will have had to apply the secret in K8s that looks something like this:

apiVersion: apps/v1
kind: Secret
metadata:
  name: storage-secret
  namespace: blogdemodeployments
type: Opaque
data:
  azurestorageaccountname: ???
  azurestorageaccountkey:  ???

where the credentials are as we used for Spark's REPL.

The Solution

This came from our friendly Azure sysadmin. Basically, it's to use OAuth, so:

spark.conf.set("fs.azure.account.auth.type",                          "OAuth")
spark.conf.set("fs.azure.account.oauth2.client.secret",               SECRET)
spark.conf.set("fs.azure.account.oauth2.client.id" ,                  APP_ID)
spark.conf.set("fs.azure.createRemoteFileSystemDuringInitialization", "true")
spark.conf.set("fs.azure.account.oauth.provider.type", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set("fs.azure.account.oauth2.client.endpoint", "https://login.microsoftonline.com/" + TENANT + "/oauth2/token")
spark.conf.set("fs.azure.account.auth.type." + accountName + ".dfs.core.windows.net", "SharedKey")
spark.conf.set("fs.azure.account.key."       + accountName + ".dfs.core.windows.net", accountKey)

where 

• The tenant is the ID of our active directory in Azure
• The app id (also known as client id) is the ID of the service principal
• The secret is something you create under the service principal which you use to authenticate (i.e. a password)

Et voila. You can now use Spark 3 to talk to Azure file systems to both read and write.

Kafka, Spark and HDFS in Docker on one Laptop

Starting Spark, HDFS and Kafka all in a Docker-ised environment is very convenient but not without its niggles. Here's what I did to run a Spark Structured Streaming app on my laptop.

Start a Kafka/ZK cluster in Docker following this link [GitHub] and for Spark/HDFS, try here [GitHub]. Note that in the Kafka/ZK config, you will have to change the value for KAFKA_ADVERTISED_HOST_NAME in docker-compose.yml to correspond to your computer each time you fire it up.     

Note that Docker creates a virtual network. You can add virtual networks to containers [SO] but you don't need to worry about that if you use docker-compose up -d. The -d switch prevents the containers output being spurged to stdout.

You can test Zookeeper is indeed up and running with:

$ echo ruok | nc localhost 2181
imok$

Note that on the host machine, you can run:

$KAFKA_HOME/bin/kafka-topics.sh --list --zookeeper localhost:2181
__consumer_offsets
test_topic

And see the internal topic __consumer_offsets has automatically been created to store offsets. The topic test_topic is something I made that we'll need later.

We can even jump onto the Kafka container and watch the logs:

$ docker exec -it kafkadocker_kafka_1 bash
bash-4.4# tail -f /opt/kafka_2.12-2.4.0/logs/server.log

Let's now see who has the Kafka port:

$ docker ps | grep 9092
02fc5122f6e2        kafkadocker_kafka                                "start-kafka.sh"         9 minutes ago       Up 9 minutes             0.0.0.0:32770->9092/tcp                                    kafkadocker_kafka_1

Note that this means that the Kafka server is listening on the host OS on port 32770 not 9092. We'll need this to tell Spark where Kafka is.

Note the Spark worker has just 1gb of heap.

$ docker exec -it dockerhadoopsparkworkbench_spark-worker_1 bash
root@a68aff72a10f:/# jps
229 Worker
1178 Jps
root@a68aff72a10f:/# cat /proc/229/cmdline
/docker-java-home/bin/java-cp/spark//conf/:/spark/jars/*:/etc/hadoop/:/opt/hadoop-2.8.0/share/hadoop/common/lib/*:/opt/hadoop-2.8.0/share/hadoop/common/*:/opt/hadoop-2.8.0/share/hadoop/hdfs/:/opt/hadoop-2.8.0/share/hadoop/hdfs/lib/*:/opt/hadoop-2.8.0/share/hadoop/hdfs/*:/opt/hadoop-2.8.0/share/hadoop/yarn/lib/*:/opt/hadoop-2.8.0/share/hadoop/yarn/*:/opt/hadoop-2.8.0/share/hadoop/mapreduce/lib/*:/opt/hadoop-2.8.0/share/hadoop/mapreduce/*:/contrib/capacity-scheduler/*.jar-Xmx1gorg.apache.spark.deploy.worker.Worker--webui-port8081spark://spark-master:7077root@a68aff72a10f:/#

If you want to execute a Spark shell, run:

$ docker exec -it spark-master /bin/bash ./spark/bin/spark-shell --master spark://spark-master:7077

Or, if you want to include a JAR:

docker run --rm -it --network dockerhadoopsparkworkbench_default --env-file ./hadoop.env -e SPARK_MASTER=spark://spark-master:7077 --volume  /home/henryp/Code/Scala/MyCode/SSSPlayground/target/:/example bde2020/spark-base:2.4.0-hadoop2.8-scala2.12 /spark/bin/spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.12:2.4.0  --jars /example/SSSPlayground-1.0-SNAPSHOT-jar-with-dependencies.jar --master spark://spark-master:7077

Docker creates a virtual network and you can see that with:

$ docker network ls
NETWORK ID          NAME                                 DRIVER              SCOPE
2f804fb10173        bridge                               bridge              local
caeab723a6c7        dockerhadoopsparkworkbench_default   bridge              local
dbb8f4df303a        host                                 host                local
b8ba799f4916        kafkadocker_default                  bridge              local
a15ee4bf8c1f        kafkasparkhadoopzk_default           bridge              local
d796a747993d        none                                 null                local

The network bridge is the default but we'll use the network in which the Spark containers sit when we deploy the local uber-jar with:

docker run --rm -it --network dockerhadoopsparkworkbench_default --env-file ./hadoop.env -e SPARK_MASTER=spark://spark-master:7077 --volume  /home/henryp/Code/Scala/MyCode/SSSPlayground/target/:/example bde2020/spark-base:2.4.0-hadoop2.8-scala2.12 /spark/bin/spark-submit --class=uk.co.odinconsultants.sssplayground.windows.ConsumeKafkaMain --master spark://spark-master:7077  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:2.4.0  /example/SSSPlayground-1.0-SNAPSHOT-jar-with-dependencies.jar 10.107.222.63:32770 test_topic /streaming_test 600000

using paths relevent to you. This creates another container called spark-base that lives for just the duration of the app. For what it's worth, the code lives here [GitHub].

This command line takes some explaining:

• We're getting docker to mount a directory with the --volume switch. 
• To have Spark streaming from Kafka you need the dependency defined with the --packages switch. This is nothing to do with Docker but is essential for Spark and Kafka to talk to each other.
• The address, 10.107.222.63, is my host OS's IP. 
• Finally, we need to tell Spark where the Kafka bootstrap servers are and this is port 32770 we saw earlier.

Pumping messages through Kafka from the host OS leads to data accumulating in HDFS and we can see that with:

$ docker exec -it dockerhadoopsparkworkbench_datanode_1  hadoop fs -du -h /
946.7 M  /streaming_test
1.9 K    /streaming_testcheckpoint
47.7 K   /tmp

So, evidently things are working.