spark/spark-with-minio.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Run spark jobs with MinIO as object storage\n",
    "\n",
    "In this Notebook we will use MinIO as object storage for Spark jobs. If you haven't already setup `spark-operator` in your Kubernetes environment follow [this](setup-spark-operator.ipynb) guide.\n",
    "\n",
    "Reading and writing Data from and to Minio using Spark is very simple once we have the right dependencies and configurations in place.\n",
    "In this post we will not be discussing the dependencies, to keep things simple we use the `openlake/spark-py:3.3.2` image that contains all the dependencies required to read and write data from Minio using Spark."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Getting Demo Data into MinIO\n",
    "We will be using the NYC Taxi dataset that is available on MinIO. You can download the dataset from [here](https://data.cityofnewyork.us/api/views/t29m-gskq/rows.csv?accessType=DOWNLOAD) which has ~112M rows and ~10GB in size. You can use any other dataset of your choice.and upload it to MinIO using the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!mc mb play/openlake\n",
    "!mc mb play/openlake/spark\n",
    "!mc mb play/openlake/spark/sample-data\n",
    "!mc cp nyc-taxi-data.csv play/openlake/spark/sample-data/nyc-taxi-data.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Sample Python Application\n",
    "Let's now read and write data from MinIO using Spark. We will use the following sample python application to do that."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile sample-code/src/main.py\n",
    "import logging\n",
    "import os\n",
    "\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import StructType, StructField, LongType, DoubleType, StringType\n",
    "\n",
    "logging.basicConfig(level=logging.INFO, format=\"%(asctime)s - %(name)s - %(levelname)s - %(message)s\")\n",
    "logger = logging.getLogger(\"MinIOSparkJob\")\n",
    "\n",
    "spark = SparkSession.builder.getOrCreate()\n",
    "\n",
    "\n",
    "def load_config(spark_context: SparkContext):\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.access.key\", os.getenv(\"AWS_ACCESS_KEY_ID\", \"openlakeuser\"))\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.secret.key\",\n",
    "                                                 os.getenv(\"AWS_SECRET_ACCESS_KEY\", \"openlakeuser\"))\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.endpoint\", os.getenv(\"ENDPOINT\", \"play.min.io:50000\"))\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.connection.ssl.enabled\", \"true\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.path.style.access\", \"true\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.attempts.maximum\", \"1\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.connection.establish.timeout\", \"5000\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.connection.timeout\", \"10000\")\n",
    "\n",
    "\n",
    "load_config(spark.sparkContext)\n",
    "\n",
    "# Define schema for NYC Taxi Data\n",
    "schema = StructType([\n",
    "    StructField('VendorID', LongType(), True),\n",
    "    StructField('tpep_pickup_datetime', StringType(), True),\n",
    "    StructField('tpep_dropoff_datetime', StringType(), True),\n",
    "    StructField('passenger_count', DoubleType(), True),\n",
    "    StructField('trip_distance', DoubleType(), True),\n",
    "    StructField('RatecodeID', DoubleType(), True),\n",
    "    StructField('store_and_fwd_flag', StringType(), True),\n",
    "    StructField('PULocationID', LongType(), True),\n",
    "    StructField('DOLocationID', LongType(), True),\n",
    "    StructField('payment_type', LongType(), True),\n",
    "    StructField('fare_amount', DoubleType(), True),\n",
    "    StructField('extra', DoubleType(), True),\n",
    "    StructField('mta_tax', DoubleType(), True),\n",
    "    StructField('tip_amount', DoubleType(), True),\n",
    "    StructField('tolls_amount', DoubleType(), True),\n",
    "    StructField('improvement_surcharge', DoubleType(), True),\n",
    "    StructField('total_amount', DoubleType(), True)])\n",
    "\n",
    "# Read CSV file from MinIO\n",
    "df = spark.read.option(\"header\", \"true\").schema(schema).csv(\n",
    "    os.getenv(\"INPUT_PATH\", \"s3a://openlake/spark/sample-data/taxi-data.csv\"))\n",
    "# Filter dataframe based on passenger_count greater than 6\n",
    "large_passengers_df = df.filter(df.passenger_count > 6)\n",
    "\n",
    "total_rows_count = df.count()\n",
    "filtered_rows_count = large_passengers_df.count()\n",
    "# File Output Committer is used to write the output to the destination (Not recommended for Production)\n",
    "large_passengers_df.write.format(\"csv\").option(\"header\", \"true\").save(\n",
    "    os.getenv(\"OUTPUT_PATH\", \"s3a://openlake-tmp/spark/nyc/taxis_small\"))\n",
    "\n",
    "logger.info(f\"Total Rows for NYC Taxi Data: {total_rows_count}\")\n",
    "logger.info(f\"Total Rows for Passenger Count > 6: {filtered_rows_count}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The above application reads the NYC Taxi dataset from MinIO and filters the rows where the passenger count is greater than 6. The filtered data is then written to MinIO."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Building the Docker Image\n",
    "We will now build the docker image that contains the above python application. You can use the following Dockerfile to build the image:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile sample-code/Dockerfile\n",
    "FROM openlake/spark-py:3.3.2\n",
    "\n",
    "USER root\n",
    "\n",
    "WORKDIR /app\n",
    "\n",
    "RUN pip3 install pyspark==3.3.2\n",
    "\n",
    "COPY src/*.py ."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can build your own docker image or use the pre-build image `openlake/sparkjob-demo:3.3.2` that is available on Docker Hub."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Deploying the MinIO Spark Application\n",
    "To read and write data from MinIO using Spark, you need to create a secret that contains the MinIO access key and secret key. You can create the secret using the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!kubectl create secret generic minio-secret \\\n",
    "    --from-literal=AWS_ACCESS_KEY_ID=openlakeuser \\\n",
    "    --from-literal=AWS_SECRET_ACCESS_KEY=openlakeuser \\\n",
    "    --from-literal=ENDPOINT=minio.openlake.io \\\n",
    "    --from-literal=AWS_REGION=us-east-1 \\\n",
    "    --namespace spark-operator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we have the secret created, we can deploy the spark application that reads and writes data from MinIO."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile sample-code/spark-job/sparkjob-minio.yml\n",
    "apiVersion: \"sparkoperator.k8s.io/v1beta2\"\n",
    "kind: SparkApplication\n",
    "metadata:\n",
    "    name: spark-minio\n",
    "    namespace: spark-operator\n",
    "spec:\n",
    "    type: Python\n",
    "    pythonVersion: \"3\"\n",
    "    mode: cluster\n",
    "    image: \"openlake/sparkjob-demo:3.3.2\"\n",
    "    imagePullPolicy: Always\n",
    "    mainApplicationFile: local:///app/main.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",
    "        memory: \"1024m\"\n",
    "        labels:\n",
    "            version: 3.3.2\n",
    "        serviceAccount: my-release-spark\n",
    "        env:\n",
    "            -   name: AWS_REGION\n",
    "                value: us-east-1\n",
    "            -   name: AWS_ACCESS_KEY_ID\n",
    "                value: openlakeuser\n",
    "            -   name: AWS_SECRET_ACCESS_KEY\n",
    "                value: openlakeuser\n",
    "    executor:\n",
    "        cores: 1\n",
    "        instances: 3\n",
    "        memory: \"1024m\"\n",
    "        labels:\n",
    "            version: 3.3.2\n",
    "        env:\n",
    "            -   name: INPUT_PATH\n",
    "                value: \"s3a://openlake/spark/sample-data/taxi-data.csv\"\n",
    "            -   name: OUTPUT_PATH\n",
    "                value: \"s3a://openlake/spark/output/taxi-data-output\"\n",
    "            -   name: AWS_REGION\n",
    "                valueFrom:\n",
    "                    secretKeyRef:\n",
    "                        name: minio-secret\n",
    "                        key: AWS_REGION\n",
    "            -   name: AWS_ACCESS_KEY_ID\n",
    "                valueFrom:\n",
    "                    secretKeyRef:\n",
    "                        name: minio-secret\n",
    "                        key: AWS_ACCESS_KEY_ID\n",
    "            -   name: AWS_SECRET_ACCESS_KEY\n",
    "                valueFrom:\n",
    "                    secretKeyRef:\n",
    "                        name: minio-secret\n",
    "                        key: AWS_SECRET_ACCESS_KEY\n",
    "            -   name: ENDPOINT\n",
    "                valueFrom:\n",
    "                    secretKeyRef:\n",
    "                        name: minio-secret\n",
    "                        key: ENDPOINT"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Above Python Spark Application YAML file contains the following configurations:\n",
    "\n",
    "* `spec.type`: The type of the application. In this case, it is a Python application.\n",
    "* `spec.pythonVersion`: The version of Python used in the application.\n",
    "* `spec.mode`: The mode of the application. In this case, it is a cluster mode application.\n",
    "* `spec.image`: The docker image that contains the application.\n",
    "* `spec.imagePullPolicy`: The image pull policy for the docker image.\n",
    "* `spec.mainApplicationFile`: The path to the main application file.\n",
    "* `spec.sparkVersion`: The version of Spark used in the application.\n",
    "* `spec.restartPolicy`: The restart policy for the application. In this case, the application will be restarted if it fails. The application will be restarted 3 times with a 10 seconds interval between each restart. If the application fails to submit, it will be restarted 5 times with a 20 seconds interval between each restart.\n",
    "* `spec.driver`: The driver configuration for the application. In this case, we are using the `my-release-spark` service account. The driver environment variables are set to read and write data from MinIO.\n",
    "* `spec.executor`: The executor configuration for the application. In this case, we are using 3 executors with 1 core and 1GB of memory each. The executor environment variables are set to read and write data from MinIO.\n",
    "\n",
    "\n",
    "You can deploy the application using the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!kubectl apply -f sample-code/spark-job/sparkjob-minio.yml"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After the application is deployed, 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": [
    "Once the application is completed, you can check the output data in MinIO. You can use the following command to list the files in the output directory:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!mc ls play/openlake/spark/output/taxi-data-output"
   ]
  },
  {
   "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 -f spark-minio-driver -n spark-operator # stop this shell once you are done"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There is also option for you to use the Spark UI to monitor the application. You can use the following command to port forward the Spark UI:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!kubectl port-forward svc/spark-minio-ui-svc 4040:4040 -n spark-operator # stop this shell once you are done browsing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In your browser, you can access the Spark UI using the following URL: `http://localhost:4040`\n",
    "\n",
    "You will see the following Spark UI:\n",
    "\n",
    "![Spark UI](./img/spark-ui.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once the application is completed, you can delete the application using the following command:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!kubectl delete sparkapplications spark-minio -n spark-operator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Deploying a Scheduled Spark Application is almost the same as deploying a normal Spark Application. The only difference is that you need to add the `spec.schedule` field to the Spark Application YAML file and the kind is `ScheduledSparkApplication`. You can save the following application as sparkjob-minio-scheduled.yaml:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile sample-code/spark-job/sparkjob-scheduled-minio.yml\n",
    "apiVersion: \"sparkoperator.k8s.io/v1beta2\"\n",
    "kind: ScheduledSparkApplication\n",
    "metadata:\n",
    "    name: spark-scheduled-minio\n",
    "    namespace: spark-operator\n",
    "spec:\n",
    "    schedule: \"@every 1h\" # Run the application every hour\n",
    "    concurrencyPolicy: Allow\n",
    "    template:\n",
    "        type: Python\n",
    "        pythonVersion: \"3\"\n",
    "        mode: cluster\n",
    "        image: \"openlake/sparkjob-demo:3.3.2\"\n",
    "        imagePullPolicy: Always\n",
    "        mainApplicationFile: local:///app/main.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",
    "            memory: \"1024m\"\n",
    "            labels:\n",
    "                version: 3.3.2\n",
    "            serviceAccount: my-release-spark\n",
    "            env:\n",
    "                -   name: AWS_REGION\n",
    "                    value: us-east-1\n",
    "                -   name: AWS_ACCESS_KEY_ID\n",
    "                    value: openlakeuser\n",
    "                -   name: AWS_SECRET_ACCESS_KEY\n",
    "                    value: openlakeuser\n",
    "        executor:\n",
    "            cores: 1\n",
    "            instances: 3\n",
    "            memory: \"1024m\"\n",
    "            labels:\n",
    "                version: 3.3.2\n",
    "            env:\n",
    "                -   name: INPUT_PATH\n",
    "                    value: \"s3a://openlake/spark/sample-data/taxi-data.csv\"\n",
    "                -   name: OUTPUT_PATH\n",
    "                    value: \"s3a://openlake/spark/output/taxi-data-output\"\n",
    "                -   name: AWS_REGION\n",
    "                    valueFrom:\n",
    "                        secretKeyRef:\n",
    "                            name: minio-secret\n",
    "                            key: AWS_REGION\n",
    "                -   name: AWS_ACCESS_KEY_ID\n",
    "                    valueFrom:\n",
    "                        secretKeyRef:\n",
    "                            name: minio-secret\n",
    "                            key: AWS_ACCESS_KEY_ID\n",
    "                -   name: AWS_SECRET_ACCESS_KEY\n",
    "                    valueFrom:\n",
    "                        secretKeyRef:\n",
    "                            name: minio-secret\n",
    "                            key: AWS_SECRET_ACCESS_KEY\n",
    "                -   name: ENDPOINT\n",
    "                    valueFrom:\n",
    "                        secretKeyRef:\n",
    "                            name: minio-secret\n",
    "                            key: ENDPOINT"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can deploy and see the results of the application in the same way as the normal Spark Application. Above Spark Application will run every hour and will write the output to the same directory in MinIO."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "All the source code for this tutorial is available in the following GitHub repository: [openlake/spark](https://github.com/minio/openlake/tree/main/spark)"
   ]
  }
 ],
 "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
}