Research: Data locality with spark-k8s

[adwk67]

As a user I want to have a plan/outline/overview of how to write spark jobs that read data from HDFS nodes within k8s (this will possibly overlap or support #117).

This is a research ticket: as a result we will able to

• understand pros/cons
• be able to estimate implementation effort
• define specific implementation ticket(s)

[adwk67]

Data locality with spark-on-kubernetes is not available by default: this is not surprising as compute and storage is handled differently in k8s compared to HDFS.

Background / Context / Links

• Build K8s with this plugin: see also
  • Talk: https://www.databricks.com/session/hdfs-on-kubernetes-lessons-learned
  • Slides: https://www.slideshare.net/databricks/hdfs-on-kuberneteslessons-learned-with-kimoon-kim
• Best practices running Spark on k8s
  • blog post here, plus associated articles
  • nothing really on data co-locality
  • recommendations to use SSD disks
• CERN event 2018
  • slides here

  • "Without data locality, network can be a
    serious problem/bottleneck"

  • ...but no recommendations (slide 50)
  • recommend native storage connectors (EOS, Kafka) for data processing (slide 49)
• Use Alluxio with Spark on K8s
  • 2020 slides
  • can run on k8s
  • Alluxio comes the UC Berkely AMPLab group, where it was developed as a data orchestrtaion layer between compute components such as Spark and storage implementations such as HDFS or S3.
  • has an active community (56 releases) with 1000+ contributors

[adwk67]

Alluxio summary

The following were tested / verified using the documentation here and here:

• quickstart local setup
• k8s cluster setup with multiple masters and an embedded journal (but no UFS / underlying file-system)
• k8s cluster setup with a single master, an embedded journal and HDFS as the UFS
  • successful tests verifying read / write operations with HDFS

Notes

• installations can use either helm or kubectl. The kubectl .yaml files can be generated directly from the helm chart using a provided script (so that the helm chart is the single source of truth: users can adapt or provide their own kubectl files). Kubectl offers more fine-grained control and was the chosen option.
• several examples require RWX PVCs for the worker volumes. - worker pods are deployed with DaemonSet, for which separate PVs cannot be defined for each worker pod. A single local or hostPath PV must be used, so that the worker will translate it to the local path. Where this is not possible - e.g. with a KinD cluster - the cluster has to be set up to allow hostPath storage (this script was used).
• the guide states that the journal must be formatted: this can be done by adding an initContainer to the alluxio-master-statefulset.yaml file:

      initContainers:
        - name: journal-format
          image: alluxio/alluxio:2.8.1
          imagePullPolicy: IfNotPresent
          securityContext:
            runAsUser: 1000
          command: ["alluxio","formatJournal"]
          envFrom:
            - configMapRef:
                name: alluxio-config
          volumeMounts:
            - name: alluxio-journal
              mountPath: /journal

• as mentioned above, the PVC definition in alluxio-worker-daemonset.yaml can be replaced with hostPath:

 persistentVolumeClaim:
   claimName: "alluxio-worker-domain-socket"
// replaced with this:
 hostPath:
   path: /tmp/kind-shared-volume/alluxio-domain
   type: DirectoryOrCreate

• the guide states that:

Alluxio Docker containers will run as non-root user alluxio with UID 1000 and GID 1000 by default.

• this means that HDFS permissions have to be changed to allow this user read/write access (or the user has to be changed)
• a compute layer (e.g. Spark) would then address alluxio URLs rather than the UFS directly
• Alluxio adds complexity and caching overhead (in that the cache has to be filled and maintained): depending on the use-case this may or may not be more desirable than improving network performance

[adwk67]

Does Spark provide HDFS data co-locality out-of-the-box?

There has been some work done on this in this repository https://github.com/apache-spark-on-k8s/spark which has been archived (the most recent work was in 2017). Some of the issues covered HDFS co-locality (e.g. apache-spark-on-k8s/spark#349, apache-spark-on-k8s/spark#350, apache-spark-on-k8s/spark#315) but these do not appear to have been merged into the main Spark code.

[lfrancke]

Thank you. For now the answer seems to be: Not easily.