spark/setup-spark-operator.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup Spark Operator on Kubernetes Cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark with Object Storage\n",
    "\n",
    "Apache Spark is an open-source, distributed computing system used for big data processing and analytics. It is designed to handle large-scale data processing with speed, efficiency, and ease of use. Spark provides a unified analytics engine for large-scale data processing, with support for multiple languages, including Java, Scala, Python, and R.\n",
    "\n",
    "The benefits of using Spark are numerous. First, it provides a high level of parallelism, which means that it can process large amounts of data quickly and efficiently across multiple nodes in a cluster. Second, Spark provides a rich set of APIs for data processing, including support for SQL queries, machine learning, graph processing, and stream processing. Third, Spark has a flexible and extensible architecture that allows developers to easily integrate with various data sources and other tools.\n",
    "\n",
    "When running Spark jobs, it is crucial to use a suitable storage system to store the input and output data. Object storage systems like MinIO provide a highly scalable and durable storage solution that can handle large amounts of data. MinIO is an open-source object storage system that can be easily deployed on-premises or in the cloud. It is S3-compatible, which means that it can be used with a wide range of tools that support the S3 API, including Spark.\n",
    "\n",
    "Using MinIO with Spark provides several benefits over traditional Hadoop Distributed File System (HDFS) or other file-based storage systems. Firstly, object storage is highly scalable and can handle large amounts of data with ease. Secondly, it provides a flexible and cost-effective storage solution that can be easily integrated with other tools and systems. Thirdly, object storage provides a highly durable storage solution, with multiple copies of data stored across multiple nodes for redundancy and fault tolerance.\n",
    "\n",
    "In summary, Apache Spark is a powerful tool for big data processing and analytics, and using object storage systems like MinIO can provide a highly scalable, durable, and flexible storage solution for running Spark jobs."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Why Spark on Kubernetes?\n",
    "Deploying Apache Spark on Kubernetes offers several advantages over deploying it standalone. Here are some reasons why:\n",
    "\n",
    "1. Resource management: Kubernetes provides powerful resource management capabilities that can help optimize resource utilization and minimize wastage. By deploying Spark on Kubernetes, you can take advantage of Kubernetes’ resource allocation and scheduling features to allocate resources to Spark jobs dynamically, based on their needs.\n",
    "2. Scalability: Kubernetes can automatically scale the resources allocated to Spark based on the workload. This means that Spark can scale up or down depending on the amount of data it needs to process, without the need for manual intervention.\n",
    "3. Fault-tolerance: Kubernetes provides built-in fault tolerance mechanisms that can help ensure the reliability of Spark clusters. If a node in the cluster fails, Kubernetes can automatically reschedule the Spark tasks to another node, ensuring that the workload is not impacted.\n",
    "4. Simplified deployment: Kubernetes offers a simplified deployment model, where you can deploy Spark using a single YAML file. This file specifies the resources required for the Spark cluster, and Kubernetes automatically handles the rest.\n",
    "5. Integration with other Kubernetes services: By deploying Spark on Kubernetes, you can take advantage of other Kubernetes services, such as monitoring and logging, to gain greater visibility into your Spark cluster's performance and health.\n",
    "\n",
    "Overall, deploying Spark on Kubernetes offers greater flexibility, scalability, and fault-tolerance than deploying it standalone. This can help organizations optimize their big data processing and analytics workloads, while reducing operational overhead and costs."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setup Spark on Kubernetes\n",
    "We will use Spark Operator to setup Spark on Kubernetes. Spark Operator is a Kubernetes controller that allows you to manage Spark applications on Kubernetes. It provides a custom resource definition (CRD) called SparkApplication, which allows you to define and run Spark applications on Kubernetes. Spark Operator also provides a web UI that allows you to easily monitor and manage Spark applications. Spark Operator is built on top of the Kubernetes Operator SDK, which is a framework for building Kubernetes operators. Spark Operator is open-source and available on [GitHub](https://github.com/GoogleCloudPlatform/spark-on-k8s-operator). It is also available as a Helm chart, which makes it easy to deploy on Kubernetes. In this tutorial, we will use the Helm chart to deploy Spark Operator on a Kubernetes cluster.\n",
    "\n",
    "Spark Operator offers various features to simplify the management of Spark applications in Kubernetes environments. These include declarative application specification and management using custom resources, automatic submission of eligible SparkApplications, native cron support for scheduled applications, and customization of Spark pods beyond native capabilities through the mutating admission webhook.\n",
    "\n",
    "Additionally, the tool supports automatic re-submission and restart of updated SparkAppliations, as well as retries of failed submissions with linear back-off. It also provides functionality to mount local Hadoop configuration as a Kubernetes ConfigMap and automatically stage local application dependencies to MinIO via [sparkctl](https://googlecloudplatform.github.io/spark-on-k8s-operator/sparkctl/). Finally, the tool supports the collection and export of application-level metrics and driver/executor metrics to Prometheus."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Prerequisites\n",
    "To follow this tutorial, you will need:\n",
    "1. A Kubernetes cluster. You can use Minikube to set up a local Kubernetes cluster on your machine.\n",
    "2. Helm, the package manager for Kubernetes. You can follow this guide to install Helm on your machine.\n",
    "3. A MinIO server running on bare metal or kubernetes. You can follow [this](https://min.io/docs/minio/linux/operations/installation.html#install-and-deploy-minio) guide to install MinIO on bare metal or [this](https://min.io/docs/minio/kubernetes/upstream/index.html) guide to install MinIO on Kubernetes or you can use [play server](https://play.min.io/) for testing purposes.\n",
    "4. A MinIO client (mc) to access the MinIO server. You can follow [this](https://docs.min.io/docs/minio-client-quickstart-guide.html) guide to install mc on your machine."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Install Spark Operator\n",
    "To install Spark Operator, you need to add the Helm repository for Spark Operator to your local Helm client. You can do this by running the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!helm repo add spark-operator https://googlecloudplatform.github.io/spark-on-k8s-operator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once the repository is added, you can install Spark Operator using the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!helm install my-release spark-operator/spark-operator \\\n",
    "--namespace spark-operator \\\n",
    "--set webhook.enable=true \\\n",
    "--set image.repository=openlake/spark-operator \\\n",
    "--set image.tag=3.3.2 \\\n",
    "--create-namespace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "You will see the following output:\n",
    "\n",
    "```bash\n",
    "LAST DEPLOYED: Mon Feb 27 19:48:33 2023\n",
    "NAMESPACE: spark-operator\n",
    "STATUS: deployed\n",
    "REVISION: 1\n",
    "TEST SUITE: None\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This command installs Spark Operator in the spark-operator namespace, and enables the mutating admission webhook. The webhook is required to enable the mounting of local Hadoop configuration as a Kubernetes ConfigMap and to configure env variables that driver and executor can use. The image repository and tag are set to the image that contains the latest version of Spark Operator. You can also use the default image repository and tag by omitting the --set image.repository and --set image.tag flags, at the time of this writing the latest Spark Operator release used 3.1.1 version of Spark whereas openlake/spark-operator used the latest 3.3.2 release of Spark. You can skip the --create-namespace flag if you already have a namespace named spark-operator. This will also monitor all the spark applications in all the namespaces.\n",
    "\n",
    "Detailed list of configuration options can be found [here](https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/blob/master/charts/spark-operator-chart/README.md#values).\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Verify Spark Operator Installation\n",
    "To verify that Spark Operator is installed successfully, you can run the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!kubectl get pods -n spark-operator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You will see the following output:\n",
    "\n",
    "```bash\n",
    "NAME                                        READY   STATUS    RESTARTS   AGE\n",
    "my-release-spark-operator-f56c4d8c4-pr857   1/1     Running   0          14m\n",
    "```\n",
    "\n",
    "Now that we have the spark operator installed, we can deploy a Spark application or Scheduled Spark application on Kubernetes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Deploy a Spark Application\n",
    "Let's try deploying one of the example simple spark application that comes with the spark operator. You can find the list of example applications [here](https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/tree/master/examples) that calcuates Pi, we will modify the [spark Pi application](https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/blob/master/examples/spark-py-pi.yaml) to use Spark 3.3.2 and run it on Kubernetes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile sample-code/spark-job/sparkjob-pi.yml\n",
    "apiVersion: \"sparkoperator.k8s.io/v1beta2\"\n",
    "kind: SparkApplication\n",
    "metadata:\n",
    "    name: pyspark-pi\n",
    "    namespace: spark-operator\n",
    "spec:\n",
    "    type: Python\n",
    "    pythonVersion: \"3\"\n",
    "    mode: cluster\n",
    "    image: \"openlake/spark-py:3.3.2\"\n",
    "    imagePullPolicy: Always\n",
    "    mainApplicationFile: local:///opt/spark/examples/src/main/python/pi.py\n",
    "    sparkVersion: \"3.3.2\"\n",
    "    restartPolicy:\n",
    "        type: OnFailure\n",
    "        onFailureRetries: 3\n",
    "        onFailureRetryInterval: 10\n",
    "        onSubmissionFailureRetries: 5\n",
    "        onSubmissionFailureRetryInterval: 20\n",
    "    driver:\n",
    "        cores: 1\n",
    "        coreLimit: \"1200m\"\n",
    "        memory: \"512m\"\n",
    "        labels:\n",
    "            version: 3.3.2\n",
    "        serviceAccount: my-release-spark\n",
    "    executor:\n",
    "        cores: 1\n",
    "        instances: 1\n",
    "        memory: \"512m\"\n",
    "        labels:\n",
    "            version: 3.3.2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The above application will calculate the value of Pi using Spark on Kubernetes. You can deploy it using the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!kubectl apply -f sample-code/spark-job/sparkjob-pi.yml"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can check the status of the application using the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!kubectl get sparkapplications -n spark-operator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You will see the following output:\n",
    "\n",
    "```bash\n",
    "NAME         STATUS      ATTEMPTS   START                  FINISH                 AGE\n",
    "pyspark-pi   COMPLETED   1          2023-02-27T15:20:29Z   2023-02-27T15:20:59Z   10m\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also check the logs of the application using the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!kubectl logs pyspark-pi-driver -n spark-operator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You will see the following output:\n",
    "\n",
    "```bash\n",
    "23/02/27 15:20:55 INFO DAGScheduler: Job 0 finished: reduce at /opt/spark/examples/src/main/python/pi.py:42, took 2.597098 s\n",
    "Pi is roughly 3.137960\n",
    "23/02/27 15:20:55 INFO SparkUI: Stopped Spark web UI at http://pyspark-pi-d73653869375fa87-driver-svc.spark-operator.svc:4040\n",
    "23/02/27 15:20:55 INFO KubernetesClusterSchedulerBackend: Shutting down all executors\n",
    "```\n",
    "\n",
    "Now that we have the simple spark application working as expected we can try to read and write data from MinIO using Spark. You can follow the guide [here](spark-with-minio.ipynb)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}