PySpark_miscellanea.nbconvert.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "7497425c",
   "metadata": {
    "colab_type": "text",
    "id": "view-in-github"
   },
   "source": [
    "<a href=\"https://colab.research.google.com/github/groda/big_data/blob/master/PySpark_miscellanea.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d6244506",
   "metadata": {
    "id": "d6244506"
   },
   "source": [
    "<a href=\"https://github.com/groda/big_data\"><div><img src=\"https://github.com/groda/big_data/blob/master/logo_bdb.png?raw=true\" align=right width=\"90\"></div></a>\n",
    "# PySpark miscellanea\n",
    "\n",
    "\n",
    "\n",
    "> _PySpark is the Python API for Apache Spark. It enables you to perform real-time, large-scale data processing in a distributed environment using Python. It also provides a PySpark shell for interactively analyzing your data._\n",
    "\n",
    "(from: [https://spark.apache.org/docs/latest/api/python/index.html](https://spark.apache.org/docs/latest/api/python/index.html))\n",
    "\n",
    "In this notebook, we showcase various tips, tricks, and insights related to PySpark.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "D5YfPFHcjDax",
   "metadata": {
    "id": "D5YfPFHcjDax"
   },
   "source": [
    "## Table of contents\n",
    "\n",
    "<div class=\"toc\"><ul class=\"toc-item\">\n",
    "<li><span><a href=\"#scrollTo=How_to_get_your_application_s_ID\">How to get your application's ID</a></span></li>\n",
    "<li><span><a href=\"#scrollTo=Default_parallelism\">Default parallelism</a></span></li>\n",
    "<li><span><a href=\"\n",
    "#scrollTo=How_to_change_PySpark_s_log_level\">How to change PySpark's log level</a></span></li>\n",
    "<li><span><a href=\"#scrollTo=Add_your_own_logging_messages\">Add your own logging messages</a></span></li>\n",
    "</ul></div>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "GGT0FAoCuefo",
   "metadata": {
    "id": "GGT0FAoCuefo"
   },
   "source": [
    "# Preliminaries\n",
    "\n",
    "The libraries needed to run this notebook. Execute this cell before any other."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "MTahPuF4uetz",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2025-01-02T13:51:36.007952Z",
     "iopub.status.busy": "2025-01-02T13:51:36.007758Z",
     "iopub.status.idle": "2025-01-02T13:51:36.120096Z",
     "shell.execute_reply": "2025-01-02T13:51:36.119454Z"
    },
    "id": "MTahPuF4uetz"
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import rand, sum, avg, stddev, expr, year\n",
    "from datetime import timedelta, date\n",
    "import os\n",
    "import subprocess"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9dfe7c0a",
   "metadata": {
    "id": "9dfe7c0a"
   },
   "source": [
    "# How to get your application's ID\n",
    "\n",
    "\n",
    "See also: [How to extract application ID from the PySpark context](https://stackoverflow.com/questions/30983226/how-to-extract-application-id-from-the-pyspark-context).\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "02dbf961",
   "metadata": {
    "id": "02dbf961"
   },
   "source": [
    "## From the Spark session\n",
    "\n",
    "What is a [Spark session](https://spark.apache.org/docs/latest/sql-getting-started.html#starting-point-sparksession)?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "0fd4ce7b",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2025-01-02T13:51:36.123596Z",
     "iopub.status.busy": "2025-01-02T13:51:36.123302Z",
     "iopub.status.idle": "2025-01-02T13:51:40.791031Z",
     "shell.execute_reply": "2025-01-02T13:51:40.790391Z"
    },
    "id": "0fd4ce7b"
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/02 13:51:39 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"My Spark App 🌟\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d02813d3",
   "metadata": {
    "id": "d02813d3"
   },
   "source": [
    "Get the session's context ([what is a Spark context?](https://spark.apache.org/docs/latest/rdd-programming-guide.html#initializing-spark) and [detailed documentation](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.SparkContext.html))."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "7d8b99e3",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 197
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:51:40.793801Z",
     "iopub.status.busy": "2025-01-02T13:51:40.793368Z",
     "iopub.status.idle": "2025-01-02T13:51:40.802446Z",
     "shell.execute_reply": "2025-01-02T13:51:40.801619Z"
    },
    "id": "7d8b99e3",
    "outputId": "e2a92282-ef3a-4f3a-ebff-f847ca66e010"
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://fv-az807-413:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.5.4</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>My Spark App 🌟</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local[*] appName=My Spark App 🌟>"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc = spark.sparkContext\n",
    "sc"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "57df230b",
   "metadata": {
    "id": "57df230b"
   },
   "source": [
    "Get `applicationId` from the context `sc`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "bed1685a",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 35
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:51:40.804921Z",
     "iopub.status.busy": "2025-01-02T13:51:40.804499Z",
     "iopub.status.idle": "2025-01-02T13:51:40.808992Z",
     "shell.execute_reply": "2025-01-02T13:51:40.808411Z"
    },
    "id": "bed1685a",
    "outputId": "18f843d7-0e76-4825-ce2d-250dd5ed1182"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'local-1735825900397'"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.applicationId"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "de33f1ce",
   "metadata": {
    "id": "de33f1ce"
   },
   "source": [
    "Or in one single step:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "f0023074",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 35
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:51:40.811042Z",
     "iopub.status.busy": "2025-01-02T13:51:40.810690Z",
     "iopub.status.idle": "2025-01-02T13:51:40.814973Z",
     "shell.execute_reply": "2025-01-02T13:51:40.814353Z"
    },
    "id": "f0023074",
    "outputId": "3b2bfc58-064d-4e41-b9ca-0bc069187bea"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'local-1735825900397'"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sparkContext.applicationId"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dc31a770",
   "metadata": {
    "id": "dc31a770"
   },
   "source": [
    "## In the PySpark shell\n",
    "\n",
    "If you're using the _PySpark shell_ (see [using the shell](https://spark.apache.org/docs/latest/rdd-programming-guide.html#initializing-spark)), `SparkContext` is created automatically and it can be accessed from the variable called `sc`."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "CdNLf1NpsvHn",
   "metadata": {
    "id": "CdNLf1NpsvHn"
   },
   "source": [
    "```\n",
    "Python 3.10.12 (main, Nov  6 2024, 20:22:13) [GCC 11.4.0] on linux\n",
    "Type \"help\", \"copyright\", \"credits\" or \"license\" for more information.\n",
    "Setting default log level to \"WARN\".\n",
    "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
    "25/01/01 11:40:51 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
    "25/01/01 11:40:52 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n",
    "Welcome to\n",
    "      ____              __\n",
    "     / __/__  ___ _____/ /__\n",
    "    _\\ \\/ _ \\/ _ `/ __/  '_/\n",
    "   /__ / .__/\\_,_/_/ /_/\\_\\   version 3.5.3\n",
    "      /_/\n",
    "\n",
    "Using Python version 3.10.12 (main, Nov  6 2024 20:22:13)\n",
    "Spark context Web UI available at http://2c6bcae43959:4041\n",
    "Spark context available as 'sc' (master = local[*], app id = local-1735731652766).\n",
    "SparkSession available as 'spark'.\n",
    ">>> sc.applicationId\n",
    "'local-1735731652766'\n",
    ">>> quit()\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "NMfLSPZXq8Ox",
   "metadata": {
    "id": "NMfLSPZXq8Ox"
   },
   "source": [
    "You can also launch a PySpark shell within the notebook environment (note: you are going to have to input your commands in a box after clicking next to the `>>>` prompt).\n",
    "\n",
    "We are using the `timeout` function (credit: [https://stackoverflow.com/a/52975118](https://stackoverflow.com/a/52975118)) to prevent the notebook from getting stuck when being executed automatically."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "N2t0ZTAvrpeR",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:51:40.817189Z",
     "iopub.status.busy": "2025-01-02T13:51:40.816828Z",
     "iopub.status.idle": "2025-01-02T13:52:00.954343Z",
     "shell.execute_reply": "2025-01-02T13:52:00.953450Z"
    },
    "id": "N2t0ZTAvrpeR",
    "outputId": "910d7219-7238-4af9-8320-15cc536aabee"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Python 3.8.18 (default, Dec 12 2024, 19:15:30) \r\n",
      "[GCC 13.2.0] on linux\r\n",
      "Type \"help\", \"copyright\", \"credits\" or \"license\" for more information.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\r\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "25/01/02 13:51:43 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "25/01/02 13:51:44 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Welcome to\r\n",
      "      ____              __\r\n",
      "     / __/__  ___ _____/ /__\r\n",
      "    _\\ \\/ _ \\/ _ `/ __/  '_/\r\n",
      "   /__ / .__/\\_,_/_/ /_/\\_\\   version 3.5.4\r\n",
      "      /_/\r\n",
      "\r\n",
      "Using Python version 3.8.18 (default, Dec 12 2024 19:15:30)\r\n",
      "Spark context Web UI available at http://fv-az807-413:4041\r\n",
      "Spark context available as 'sc' (master = local[*], app id = local-1735825904427).\r\n",
      "SparkSession available as 'spark'.\r\n"
     ]
    }
   ],
   "source": [
    "!timeout 20 pyspark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "UqaM4j78tTlG",
   "metadata": {
    "id": "UqaM4j78tTlG"
   },
   "source": [
    "Close the Spark session with `stop()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "PoVb-kEdtTvj",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:00.957271Z",
     "iopub.status.busy": "2025-01-02T13:52:00.956766Z",
     "iopub.status.idle": "2025-01-02T13:52:01.189279Z",
     "shell.execute_reply": "2025-01-02T13:52:01.188597Z"
    },
    "id": "PoVb-kEdtTvj"
   },
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "44ece7a4",
   "metadata": {
    "id": "44ece7a4"
   },
   "source": [
    "# Default parallelism"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "238722de",
   "metadata": {
    "id": "238722de"
   },
   "source": [
    "## What is `spark.default.parallelism`?\n",
    "\n",
    "This property determines the default number of chunks in which an RDD ([Resilient Distributed Dataset](https://spark.apache.org/docs/latest/rdd-programming-guide.html#resilient-distributed-datasets-rdds)) is partitioned. This, in turn, affects how many tasks are executed concurrently."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "jmH1o6CpRR0Y",
   "metadata": {
    "id": "jmH1o6CpRR0Y"
   },
   "source": [
    "Unless specified by the user, the default value of `default.parallelism` is set based on the _cluster manager_:\n",
    " - in standalone mode it is equal to the number of (virtual) cores on the local machine\n",
    " - in Mesos: 8\n",
    " - for YARN, Kubernetes: total number of cores on all executor nodes or 2, whichever is larger\n",
    "\n",
    " (see [Spark configuration/Execution behavior](https://spark.apache.org/docs/latest/configuration.html#execution-behavior))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "pimhDh1nk4Q8",
   "metadata": {
    "id": "pimhDh1nk4Q8"
   },
   "source": [
    "## Get or set default parallelism"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "880e6ada",
   "metadata": {
    "id": "880e6ada"
   },
   "source": [
    "Create a [Spark session](https://spark.apache.org/docs/latest/sql-getting-started.html#starting-point-sparksession)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "451653da",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:01.192263Z",
     "iopub.status.busy": "2025-01-02T13:52:01.191784Z",
     "iopub.status.idle": "2025-01-02T13:52:01.294018Z",
     "shell.execute_reply": "2025-01-02T13:52:01.293192Z"
    },
    "id": "451653da"
   },
   "outputs": [],
   "source": [
    "spark = SparkSession \\\n",
    "        .builder \\\n",
    "        .appName(\"Default Parallelism 🧵🧵\") \\\n",
    "        .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d2fe39a2",
   "metadata": {
    "id": "d2fe39a2"
   },
   "source": [
    "Show the value of `defaultParallelism`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "f221e3d1",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:01.296976Z",
     "iopub.status.busy": "2025-01-02T13:52:01.296586Z",
     "iopub.status.idle": "2025-01-02T13:52:01.301254Z",
     "shell.execute_reply": "2025-01-02T13:52:01.300710Z"
    },
    "id": "f221e3d1",
    "outputId": "270bf66b-d3b2-4a8e-ca09-8925823ecd37"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sparkContext.defaultParallelism"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "69b3994c",
   "metadata": {
    "id": "69b3994c"
   },
   "source": [
    "To change a property it's necessary to stop and start a new context/session, you can't just change the configuration on an existing session!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "4cf47f42",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:01.303441Z",
     "iopub.status.busy": "2025-01-02T13:52:01.303112Z",
     "iopub.status.idle": "2025-01-02T13:52:01.975826Z",
     "shell.execute_reply": "2025-01-02T13:52:01.974538Z"
    },
    "id": "4cf47f42"
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/02 13:52:01 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession \\\n",
    "        .builder \\\n",
    "        .config(\"spark.default.parallelism\", 4) \\\n",
    "        .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c6725687",
   "metadata": {
    "id": "c6725687"
   },
   "source": [
    "Default parallelism hasn't changed!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "6b592d7e",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:01.979613Z",
     "iopub.status.busy": "2025-01-02T13:52:01.978468Z",
     "iopub.status.idle": "2025-01-02T13:52:01.984886Z",
     "shell.execute_reply": "2025-01-02T13:52:01.984129Z"
    },
    "id": "6b592d7e",
    "outputId": "5804fc3b-24d5-4aa7-d7ad-bb68a2dcace5"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sparkContext.defaultParallelism"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "g3r7LB2yHDiW",
   "metadata": {
    "id": "g3r7LB2yHDiW"
   },
   "source": [
    "Same with [`SparkSession.conf.set`](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.SparkSession.conf.html#pyspark.sql.SparkSession.conf):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "pfBzJnnzGm5Q",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:01.987819Z",
     "iopub.status.busy": "2025-01-02T13:52:01.987432Z",
     "iopub.status.idle": "2025-01-02T13:52:01.995578Z",
     "shell.execute_reply": "2025-01-02T13:52:01.994114Z"
    },
    "id": "pfBzJnnzGm5Q",
    "outputId": "91443e08-118e-4849-b793-0f1905a54c5e"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.conf.set(\"spark.default.parallelism\", 3)\n",
    "spark.sparkContext.defaultParallelism"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "49e168d0",
   "metadata": {
    "id": "49e168d0"
   },
   "source": [
    "Stop and start session anew."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "da0512cd",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:01.998872Z",
     "iopub.status.busy": "2025-01-02T13:52:01.998591Z",
     "iopub.status.idle": "2025-01-02T13:52:02.362670Z",
     "shell.execute_reply": "2025-01-02T13:52:02.361964Z"
    },
    "id": "da0512cd"
   },
   "outputs": [],
   "source": [
    "spark.stop()\n",
    "spark = SparkSession \\\n",
    "        .builder \\\n",
    "        .appName(\"Default Parallelism 🧵🧵🧵🧵\") \\\n",
    "        .config(\"spark.default.parallelism\", 4) \\\n",
    "        .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "bba24e5e",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:02.365335Z",
     "iopub.status.busy": "2025-01-02T13:52:02.365113Z",
     "iopub.status.idle": "2025-01-02T13:52:02.370062Z",
     "shell.execute_reply": "2025-01-02T13:52:02.369593Z"
    },
    "id": "bba24e5e",
    "outputId": "63f25982-bc1d-4d8a-df83-7b716a6b9972"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sparkContext.defaultParallelism"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1ef9f91f",
   "metadata": {
    "id": "1ef9f91f"
   },
   "source": [
    "Great! Now the context has been changed (and also the applications's name has been updated)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "0e2fbc63",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 0
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:02.372419Z",
     "iopub.status.busy": "2025-01-02T13:52:02.372219Z",
     "iopub.status.idle": "2025-01-02T13:52:02.377837Z",
     "shell.execute_reply": "2025-01-02T13:52:02.376940Z"
    },
    "id": "0e2fbc63",
    "outputId": "18a059dc-a8b9-41d6-d775-536f3d0e03f5"
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://fv-az807-413:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.5.4</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Default Parallelism 🧵🧵🧵🧵</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local[*] appName=Default Parallelism 🧵🧵🧵🧵>"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2e5p9uyfFfx0",
   "metadata": {
    "id": "2e5p9uyfFfx0"
   },
   "source": [
    "The reason why you cannot change a \"running\" context is that\n",
    "\n",
    "> _Once a `SparkConf` object is passed to Spark, it is cloned and can no longer be modified by the user._\n",
    "\n",
    "(see [`SparkConf`](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.SparkConf.html#pyspark.SparkConf) in the PySpark API documentation)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "pnIik2A5lD8b",
   "metadata": {
    "id": "pnIik2A5lD8b"
   },
   "source": [
    "## Example of how parallelism can influence the runtime of computation\n",
    "\n",
    "Of course parallelism is ultimately limited by the number of available virtual cores but if your cluster has sufficient resources, increasing the value of `default.parallelism` has the potential of speeding up computations.\n",
    "\n",
    "The standard Google Colab notebook has two cores, so $2$ is the maximum parallelization that can be achieved."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "y_IA5tcD0wql",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:02.381012Z",
     "iopub.status.busy": "2025-01-02T13:52:02.380819Z",
     "iopub.status.idle": "2025-01-02T13:52:02.384254Z",
     "shell.execute_reply": "2025-01-02T13:52:02.383641Z"
    },
    "id": "y_IA5tcD0wql",
    "outputId": "82e77377-745f-4ce8-a0c6-bba2ea6c8f42"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of cores: 4\n"
     ]
    }
   ],
   "source": [
    "print(f\"Number of cores: {os.cpu_count()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "sl6PQsPgvUr-",
   "metadata": {
    "id": "sl6PQsPgvUr-"
   },
   "source": [
    "We are going to create a DataFrame with random numbers using `range` and then compute the sum, average, standard deviation, and median of all values."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "1Udn9tjGvU7J",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:02.386549Z",
     "iopub.status.busy": "2025-01-02T13:52:02.386020Z",
     "iopub.status.idle": "2025-01-02T13:52:03.846243Z",
     "shell.execute_reply": "2025-01-02T13:52:03.845627Z"
    },
    "id": "1Udn9tjGvU7J"
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/02 13:52:02 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.\n"
     ]
    }
   ],
   "source": [
    "# Create SparkSession\n",
    "spark = SparkSession.builder.appName(\"Parallelism Demo ⚙️⚙️\").getOrCreate()\n",
    "\n",
    "# Create a DataFrame with random numbers\n",
    "df = spark.range(10**4).withColumn(\"value\", rand())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "u5iuCUMPD2Jv",
   "metadata": {
    "id": "u5iuCUMPD2Jv"
   },
   "source": [
    "Define an aggregation function that computes the sum of all numbers."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "2HZ-1ENfD1Ax",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:03.848766Z",
     "iopub.status.busy": "2025-01-02T13:52:03.848439Z",
     "iopub.status.idle": "2025-01-02T13:52:03.852309Z",
     "shell.execute_reply": "2025-01-02T13:52:03.851774Z"
    },
    "id": "2HZ-1ENfD1Ax"
   },
   "outputs": [],
   "source": [
    "def aggregate_data(data_frame):\n",
    "    result = data_frame.groupBy().agg(\n",
    "        sum(\"value\").alias(\"total_value\"),\n",
    "        avg(\"value\").alias(\"average_value\"),\n",
    "        stddev(\"value\").alias(\"std_deviation\"),\n",
    "        expr(\"percentile_approx(value, 0.5)\").alias(\"median\")  # Approximate median\n",
    "    )\n",
    "    return result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "nTXukfqEzEiA",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:03.854265Z",
     "iopub.status.busy": "2025-01-02T13:52:03.854016Z",
     "iopub.status.idle": "2025-01-02T13:52:06.232319Z",
     "shell.execute_reply": "2025-01-02T13:52:06.231521Z"
    },
    "id": "nTXukfqEzEiA",
    "outputId": "d2383d0d-d2fe-4c47-9557-1cc41e9f47b7"
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 0:>                                                          (0 + 4) / 4]\r"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[Row(total_value=5008.810133549655, average_value=0.5008810133549655, std_deviation=0.28797886773772013, median=0.4984467182791408)]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "aggregate_data(df).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "RVQkbt6_1L8z",
   "metadata": {
    "id": "RVQkbt6_1L8z"
   },
   "source": [
    "Now run the same computation with different parallelism settings (but note that you won't be able to appreciate the effect of parallelism unless you try this on a system with more than 2 CPUs!).\n",
    "\n",
    "⚠️ Warning: the following computation might take a couple of minutes to run since it performs some calculations on a dataframe with $10^8$ rows and on Colab you get a maximum parallelism of $2$."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "OjUFcn0PDpb5",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:52:06.234740Z",
     "iopub.status.busy": "2025-01-02T13:52:06.234520Z",
     "iopub.status.idle": "2025-01-02T13:53:33.847415Z",
     "shell.execute_reply": "2025-01-02T13:53:33.846448Z"
    },
    "id": "OjUFcn0PDpb5",
    "outputId": "e1effc38-0c82-49bc-9baa-00195694cfa9"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Parallelism: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 0:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 12.7 ms, sys: 10.5 ms, total: 23.2 ms\n",
      "Wall time: 35.9 s\n",
      "\n",
      "Parallelism: 2\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 0:>                                                          (0 + 2) / 2]\r"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 0:=============================>                             (1 + 1) / 2]\r"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 10.4 ms, sys: 6.24 ms, total: 16.6 ms\n",
      "Wall time: 19.5 s\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Parallelism: 4\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 0:>                                                          (0 + 4) / 4]\r"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 14.3 ms, sys: 3.49 ms, total: 17.8 ms\n",
      "Wall time: 15.5 s\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Parallelism: 8\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 0:>                                                          (0 + 4) / 8]\r"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 0:=============================>                             (4 + 4) / 8]\r"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 0:====================================>                      (5 + 3) / 8]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 12.7 ms, sys: 5.15 ms, total: 17.9 ms\n",
      "Wall time: 15.1 s\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "for parallelism in [1, 2, 4, 8]:\n",
    "  # Create SparkSession\n",
    "  spark.stop()\n",
    "  spark = SparkSession.builder.appName(f\"Default Parallelism {'🧵'*parallelism}\") \\\n",
    "                      .config(\"spark.default.parallelism\", parallelism) \\\n",
    "                      .getOrCreate()\n",
    "  # Create a DataFrame with random numbers\n",
    "  df = spark.range(10**8).withColumn(\"value\", rand())\n",
    "\n",
    "  print(f\"Parallelism: {spark.sparkContext.getConf().get('spark.default.parallelism')}\")\n",
    "  %time aggregate_data(df).collect() # Trigger the computation\n",
    "  print(\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "F_di5wA-QHdn",
   "metadata": {
    "id": "F_di5wA-QHdn"
   },
   "source": [
    "I ran the same code on my laptop with $8$ CPUs and got better results where one can see how increasing parallelism reduces the total runtime.\n",
    "\n",
    "```\n",
    "Parallelism: 1\n",
    "CPU times: user 8.22 ms, sys: 3.96 ms, total: 12.2 ms                           \n",
    "Wall time: 42.4 s\n",
    "\n",
    "Parallelism: 2\n",
    "CPU times: user 4.96 ms, sys: 1.97 ms, total: 6.94 ms                           \n",
    "Wall time: 23 s\n",
    "\n",
    "Parallelism: 4\n",
    "CPU times: user 4.27 ms, sys: 1.66 ms, total: 5.93 ms                           \n",
    "Wall time: 16 s\n",
    "\n",
    "Parallelism: 8\n",
    "CPU times: user 4.58 ms, sys: 1.78 ms, total: 6.36 ms                           \n",
    "Wall time: 15 s\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ho2KnZNrxyea",
   "metadata": {
    "id": "ho2KnZNrxyea"
   },
   "source": [
    "# How to change PySpark's log level\n",
    "\n",
    "Log levels in PySpark sorted from the most verbose to the least are:\n",
    "\n",
    "\n",
    "* ALL\n",
    "* TRACE\n",
    "* DEBUG\n",
    "* INFO\n",
    "* WARN\n",
    "* ERROR\n",
    "* FATAL\n",
    "* OFF\n",
    "\n",
    "\n",
    "See [https://spark.apache.org/docs/.../api/pyspark.SparkContext.setLogLevel.html](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.SparkContext.setLogLevel.html).\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "xwXJJLUZx6Tq",
   "metadata": {
    "id": "xwXJJLUZx6Tq"
   },
   "source": [
    "## In the PySpark shell\n",
    "\n",
    "To change the log level to \"INFO\" in the PySpark shell just enter:\n",
    "\n",
    "    sc.setLogLevel(\"INFO\")\n",
    "\n",
    "⚠️ Anything log level above \"INFO\" is extermely verbose, so be prepared for a lot of output!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "lnIAoX_1TG_k",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:53:33.850884Z",
     "iopub.status.busy": "2025-01-02T13:53:33.850566Z",
     "iopub.status.idle": "2025-01-02T13:57:33.994673Z",
     "shell.execute_reply": "2025-01-02T13:57:33.993306Z"
    },
    "id": "lnIAoX_1TG_k",
    "outputId": "a56bf0e8-a500-434f-9996-6deab2d08a93"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Python 3.8.18 (default, Dec 12 2024, 19:15:30) \r\n",
      "[GCC 13.2.0] on linux\r\n",
      "Type \"help\", \"copyright\", \"credits\" or \"license\" for more information.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\r\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "25/01/02 13:53:36 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "25/01/02 13:53:36 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Welcome to\r\n",
      "      ____              __\r\n",
      "     / __/__  ___ _____/ /__\r\n",
      "    _\\ \\/ _ \\/ _ `/ __/  '_/\r\n",
      "   /__ / .__/\\_,_/_/ /_/\\_\\   version 3.5.4\r\n",
      "      /_/\r\n",
      "\r\n",
      "Using Python version 3.8.18 (default, Dec 12 2024 19:15:30)\r\n",
      "Spark context Web UI available at http://fv-az807-413:4041\r\n",
      "Spark context available as 'sc' (master = local[*], app id = local-1735826017048).\r\n",
      "SparkSession available as 'spark'.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "25/01/02 13:57:33 WARN HeartbeatReceiver: Removing executor driver with no recent heartbeats: 236646 ms exceeds timeout 120000 ms\r\n"
     ]
    }
   ],
   "source": [
    "!timeout 240 pyspark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "kIsl8s5jWG5S",
   "metadata": {
    "id": "kIsl8s5jWG5S"
   },
   "source": [
    "```\n",
    "from pyspark.sql.functions import year\n",
    "from datetime import timedelta, date\n",
    "\n",
    "df = spark.createDataFrame([\n",
    "    {\"date\": date.today(), \"value\": 2.1},\n",
    "    {\"date\": date.today() + timedelta(days=1), \"value\": 1.9},\n",
    "    {\"date\": date.today() + timedelta(days=2), \"value\": 2.3},\n",
    "    {\"date\": date.today() - timedelta(days=365*5), \"value\": 3.0},\n",
    "])\n",
    "\n",
    "sc = spark.sparkContext\n",
    "sc.setLogLevel(\"INFO\")\n",
    "df.groupBy(year(\"date\")).avg().show()\n",
    "sc.setLogLevel(\"WARN\")\n",
    "df.groupBy(year(\"date\")).avg().show()\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "r2VALwrWV5Tw",
   "metadata": {
    "id": "r2VALwrWV5Tw"
   },
   "source": [
    "## In a PySpark script\n",
    "\n",
    "After finding `$SPARK_HOME`, we are going to create a Log4J configuration file and finally run a PySpark script showcasing different log levels.\n",
    "\n",
    "See also: [PySpark on Google Colab](https://github.com/groda/big_data/blob/master/PySpark_On_Google_Colab.ipynb)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "2cppX3R-UdvB",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:57:33.998686Z",
     "iopub.status.busy": "2025-01-02T13:57:33.998277Z",
     "iopub.status.idle": "2025-01-02T13:57:34.144722Z",
     "shell.execute_reply": "2025-01-02T13:57:34.143850Z"
    },
    "id": "2cppX3R-UdvB",
    "outputId": "722fa43c-36e0-4ca5-c983-75c83ab68354"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/opt/hostedtoolcache/Python/3.8.18/x64/lib/python3.8/site-packages/pyspark\r\n"
     ]
    }
   ],
   "source": [
    "!find_spark_home.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "BUhX4x1tWxtC",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:57:34.147453Z",
     "iopub.status.busy": "2025-01-02T13:57:34.147232Z",
     "iopub.status.idle": "2025-01-02T13:57:34.174422Z",
     "shell.execute_reply": "2025-01-02T13:57:34.173696Z"
    },
    "id": "BUhX4x1tWxtC",
    "outputId": "caa13b29-5511-4f86-f1c8-0026a1efb983"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Output of find_spark_home.py: /opt/hostedtoolcache/Python/3.8.18/x64/lib/python3.8/site-packages/pyspark\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Run the script and capture its output\n",
    "result = subprocess.run([\"find_spark_home.py\"], capture_output=True, text=True)\n",
    "\n",
    "# Print or use the captured output\n",
    "print(\"Output of find_spark_home.py:\", result.stdout)\n",
    "\n",
    "# set SPARK_HOME environment variable\n",
    "os.environ['SPARK_HOME'] = result.stdout.strip()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "FAI_9pw1qrbJ",
   "metadata": {
    "id": "FAI_9pw1qrbJ"
   },
   "source": [
    "Now the variable `SPARK_HOME` is set."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "sQHZdV5EW_DL",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:57:34.177199Z",
     "iopub.status.busy": "2025-01-02T13:57:34.176511Z",
     "iopub.status.idle": "2025-01-02T13:57:34.302550Z",
     "shell.execute_reply": "2025-01-02T13:57:34.301829Z"
    },
    "id": "sQHZdV5EW_DL",
    "outputId": "1e02f290-2d99-46a4-a8c6-95c109912923"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/opt/hostedtoolcache/Python/3.8.18/x64/lib/python3.8/site-packages/pyspark\r\n"
     ]
    }
   ],
   "source": [
    "!echo $SPARK_HOME"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "HH7_bAv6qe14",
   "metadata": {
    "id": "HH7_bAv6qe14"
   },
   "source": [
    "Create a `log4j2.properties` file in Spark's configuration directory."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "W96FZEFcWcPo",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2025-01-02T13:57:34.305230Z",
     "iopub.status.busy": "2025-01-02T13:57:34.304806Z",
     "iopub.status.idle": "2025-01-02T13:57:34.319666Z",
     "shell.execute_reply": "2025-01-02T13:57:34.318840Z"
    },
    "id": "W96FZEFcWcPo"
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "# create conf directory\n",
    "# with the option -p mkdir won't complain if the folder already exists\n",
    "mkdir -p $SPARK_HOME/conf\n",
    "\n",
    "# populate log4j2.properties file\n",
    "FILE=$SPARK_HOME/conf/log4j2.properties\n",
    "\n",
    "# read about heredocs: https://tldp.org/LDP/abs/html/here-docs.html\n",
    "cat> $FILE <<🤖\n",
    "status = warn\n",
    "\n",
    "appender.console.type = Console\n",
    "appender.console.name = STDOUT\n",
    "appender.console.target = SYSTEM_ERR\n",
    "\n",
    "rootLogger.level = warn\n",
    "rootLogger.appenderRef.stdout.ref = STDOUT\n",
    "\n",
    "# formatting\n",
    "appender.console.layout.type = PatternLayout\n",
    "appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n\n",
    "🤖"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "gmKvfBl3XkzO",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:57:34.321927Z",
     "iopub.status.busy": "2025-01-02T13:57:34.321544Z",
     "iopub.status.idle": "2025-01-02T13:57:34.326780Z",
     "shell.execute_reply": "2025-01-02T13:57:34.326128Z"
    },
    "id": "gmKvfBl3XkzO",
    "outputId": "d941f999-3819-41d7-9efa-ca50d447423b"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing my_app.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile my_app.py\n",
    "from pyspark.sql import SparkSession\n",
    "from datetime import timedelta, date\n",
    "from pyspark.sql.functions import year\n",
    "import logging\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    logging.basicConfig(level=logging.INFO)\n",
    "    logger = logging.getLogger(__name__)\n",
    "\n",
    "    logger.info(\"Creating Spark session\")\n",
    "    spark = SparkSession.builder.appName(\"Logging levels 📝\").getOrCreate()\n",
    "\n",
    "    df = spark.createDataFrame([\n",
    "        {\"date\": date.today(), \"value\": 2.1},\n",
    "        {\"date\": date.today() + timedelta(days=1), \"value\": 1.9},\n",
    "        {\"date\": date.today() + timedelta(days=2), \"value\": 2.3},\n",
    "        {\"date\": date.today() - timedelta(days=365*5), \"value\": 3.0},\n",
    "    ])\n",
    "\n",
    "    df.groupBy(year(\"date\")).avg().show()\n",
    "    sc = spark.sparkContext\n",
    "    logger.error(\"Setting log level to INFO\")\n",
    "    sc.setLogLevel(\"INFO\")\n",
    "    df.groupBy(year(\"date\")).avg().show()\n",
    "    spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "qwNPA_BN5sQs",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:57:34.329003Z",
     "iopub.status.busy": "2025-01-02T13:57:34.328638Z",
     "iopub.status.idle": "2025-01-02T13:57:44.352824Z",
     "shell.execute_reply": "2025-01-02T13:57:44.352078Z"
    },
    "id": "qwNPA_BN5sQs",
    "outputId": "cf91872b-0eb7-40d5-8173-cd126680f1ea"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "INFO:__main__:Creating Spark session\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:36 WARN  NativeCodeLoader:60 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:37 WARN  Utils:72 - Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+\r\n",
      "|year(date)|avg(value)|\r\n",
      "+----------+----------+\r\n",
      "|      2025|       2.1|\r\n",
      "|      2020|       3.0|\r\n",
      "+----------+----------+\r\n",
      "\r\n",
      "ERROR:__main__:Setting log level to INFO\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:42 INFO  DAGScheduler:60 - Registering RDD 11 (showString at NativeMethodAccessorImpl.java:0) as input to shuffle 1\r\n",
      "2025-01-02 13:57:42 INFO  DAGScheduler:60 - Got map stage job 2 (showString at NativeMethodAccessorImpl.java:0) with 4 output partitions\r\n",
      "2025-01-02 13:57:42 INFO  DAGScheduler:60 - Final stage: ShuffleMapStage 3 (showString at NativeMethodAccessorImpl.java:0)\r\n",
      "2025-01-02 13:57:42 INFO  DAGScheduler:60 - Parents of final stage: List()\r\n",
      "2025-01-02 13:57:42 INFO  DAGScheduler:60 - Missing parents: List()\r\n",
      "2025-01-02 13:57:42 INFO  DAGScheduler:60 - Submitting ShuffleMapStage 3 (MapPartitionsRDD[11] at showString at NativeMethodAccessorImpl.java:0), which has no missing parents\r\n",
      "2025-01-02 13:57:42 INFO  MemoryStore:60 - Block broadcast_2 stored as values in memory (estimated size 41.0 KiB, free 434.4 MiB)\r\n",
      "2025-01-02 13:57:42 INFO  MemoryStore:60 - Block broadcast_2_piece0 stored as bytes in memory (estimated size 19.2 KiB, free 434.3 MiB)\r\n",
      "2025-01-02 13:57:42 INFO  BlockManagerInfo:60 - Added broadcast_2_piece0 in memory on fv-az807-413:34879 (size: 19.2 KiB, free: 434.4 MiB)\r\n",
      "2025-01-02 13:57:42 INFO  SparkContext:60 - Created broadcast 2 from broadcast at DAGScheduler.scala:1585\r\n",
      "2025-01-02 13:57:42 INFO  DAGScheduler:60 - Submitting 4 missing tasks from ShuffleMapStage 3 (MapPartitionsRDD[11] at showString at NativeMethodAccessorImpl.java:0) (first 15 tasks are for partitions Vector(0, 1, 2, 3))\r\n",
      "2025-01-02 13:57:42 INFO  TaskSchedulerImpl:60 - Adding task set 3.0 with 4 tasks resource profile 0\r\n",
      "2025-01-02 13:57:42 INFO  TaskSetManager:60 - Starting task 0.0 in stage 3.0 (TID 5) (fv-az807-413, executor driver, partition 0, PROCESS_LOCAL, 9007 bytes) \r\n",
      "2025-01-02 13:57:42 INFO  TaskSetManager:60 - Starting task 1.0 in stage 3.0 (TID 6) (fv-az807-413, executor driver, partition 1, PROCESS_LOCAL, 9007 bytes) \r\n",
      "2025-01-02 13:57:42 INFO  TaskSetManager:60 - Starting task 2.0 in stage 3.0 (TID 7) (fv-az807-413, executor driver, partition 2, PROCESS_LOCAL, 9007 bytes) \r\n",
      "2025-01-02 13:57:42 INFO  TaskSetManager:60 - Starting task 3.0 in stage 3.0 (TID 8) (fv-az807-413, executor driver, partition 3, PROCESS_LOCAL, 9007 bytes) \r\n",
      "2025-01-02 13:57:42 INFO  Executor:60 - Running task 1.0 in stage 3.0 (TID 6)\r\n",
      "2025-01-02 13:57:42 INFO  Executor:60 - Running task 0.0 in stage 3.0 (TID 5)\r\n",
      "2025-01-02 13:57:42 INFO  Executor:60 - Running task 2.0 in stage 3.0 (TID 7)\r\n",
      "2025-01-02 13:57:42 INFO  Executor:60 - Running task 3.0 in stage 3.0 (TID 8)\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:43 INFO  PythonRunner:60 - Times: total = 43, boot = -1451, init = 1494, finish = 0\r\n",
      "2025-01-02 13:57:43 INFO  PythonRunner:60 - Times: total = 44, boot = -1454, init = 1498, finish = 0\r\n",
      "2025-01-02 13:57:43 INFO  PythonRunner:60 - Times: total = 44, boot = -1471, init = 1515, finish = 0\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:43 INFO  PythonRunner:60 - Times: total = 45, boot = -1461, init = 1506, finish = 0\r\n",
      "2025-01-02 13:57:43 INFO  Executor:60 - Finished task 3.0 in stage 3.0 (TID 8). 2839 bytes result sent to driver\r\n",
      "2025-01-02 13:57:43 INFO  Executor:60 - Finished task 0.0 in stage 3.0 (TID 5). 2839 bytes result sent to driver\r\n",
      "2025-01-02 13:57:43 INFO  Executor:60 - Finished task 1.0 in stage 3.0 (TID 6). 2839 bytes result sent to driver\r\n",
      "2025-01-02 13:57:43 INFO  Executor:60 - Finished task 2.0 in stage 3.0 (TID 7). 2839 bytes result sent to driver\r\n",
      "2025-01-02 13:57:43 INFO  TaskSetManager:60 - Finished task 1.0 in stage 3.0 (TID 6) in 91 ms on fv-az807-413 (executor driver) (1/4)\r\n",
      "2025-01-02 13:57:43 INFO  TaskSetManager:60 - Finished task 3.0 in stage 3.0 (TID 8) in 91 ms on fv-az807-413 (executor driver) (2/4)\r\n",
      "2025-01-02 13:57:43 INFO  TaskSetManager:60 - Finished task 0.0 in stage 3.0 (TID 5) in 94 ms on fv-az807-413 (executor driver) (3/4)\r\n",
      "2025-01-02 13:57:43 INFO  TaskSetManager:60 - Finished task 2.0 in stage 3.0 (TID 7) in 93 ms on fv-az807-413 (executor driver) (4/4)\r\n",
      "2025-01-02 13:57:43 INFO  TaskSchedulerImpl:60 - Removed TaskSet 3.0, whose tasks have all completed, from pool \r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - ShuffleMapStage 3 (showString at NativeMethodAccessorImpl.java:0) finished in 0.111 s\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - looking for newly runnable stages\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - running: Set()\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - waiting: Set()\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - failed: Set()\r\n",
      "2025-01-02 13:57:43 INFO  ShufflePartitionsUtil:60 - For shuffle(1), advisory target size: 67108864, actual target size 1048576, minimum partition size: 1048576\r\n",
      "2025-01-02 13:57:43 INFO  HashAggregateExec:60 - spark.sql.codegen.aggregate.map.twolevel.enabled is set to true, but current version of codegened fast hashmap does not support this aggregate.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:43 INFO  SparkContext:60 - Starting job: showString at NativeMethodAccessorImpl.java:0\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - Got job 3 (showString at NativeMethodAccessorImpl.java:0) with 1 output partitions\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - Final stage: ResultStage 5 (showString at NativeMethodAccessorImpl.java:0)\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - Parents of final stage: List(ShuffleMapStage 4)\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - Missing parents: List()\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - Submitting ResultStage 5 (MapPartitionsRDD[14] at showString at NativeMethodAccessorImpl.java:0), which has no missing parents\r\n",
      "2025-01-02 13:57:43 INFO  MemoryStore:60 - Block broadcast_3 stored as values in memory (estimated size 45.0 KiB, free 434.3 MiB)\r\n",
      "2025-01-02 13:57:43 INFO  MemoryStore:60 - Block broadcast_3_piece0 stored as bytes in memory (estimated size 21.2 KiB, free 434.3 MiB)\r\n",
      "2025-01-02 13:57:43 INFO  BlockManagerInfo:60 - Added broadcast_3_piece0 in memory on fv-az807-413:34879 (size: 21.2 KiB, free: 434.4 MiB)\r\n",
      "2025-01-02 13:57:43 INFO  SparkContext:60 - Created broadcast 3 from broadcast at DAGScheduler.scala:1585\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - Submitting 1 missing tasks from ResultStage 5 (MapPartitionsRDD[14] at showString at NativeMethodAccessorImpl.java:0) (first 15 tasks are for partitions Vector(0))\r\n",
      "2025-01-02 13:57:43 INFO  TaskSchedulerImpl:60 - Adding task set 5.0 with 1 tasks resource profile 0\r\n",
      "2025-01-02 13:57:43 INFO  TaskSetManager:60 - Starting task 0.0 in stage 5.0 (TID 9) (fv-az807-413, executor driver, partition 0, NODE_LOCAL, 8999 bytes) \r\n",
      "2025-01-02 13:57:43 INFO  Executor:60 - Running task 0.0 in stage 5.0 (TID 9)\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:43 INFO  ShuffleBlockFetcherIterator:60 - Getting 4 (288.0 B) non-empty blocks including 4 (288.0 B) local and 0 (0.0 B) host-local and 0 (0.0 B) push-merged-local and 0 (0.0 B) remote blocks\r\n",
      "2025-01-02 13:57:43 INFO  ShuffleBlockFetcherIterator:60 - Started 0 remote fetches in 1 ms\r\n",
      "2025-01-02 13:57:43 INFO  Executor:60 - Finished task 0.0 in stage 5.0 (TID 9). 5273 bytes result sent to driver\r\n",
      "2025-01-02 13:57:43 INFO  TaskSetManager:60 - Finished task 0.0 in stage 5.0 (TID 9) in 28 ms on fv-az807-413 (executor driver) (1/1)\r\n",
      "2025-01-02 13:57:43 INFO  TaskSchedulerImpl:60 - Removed TaskSet 5.0, whose tasks have all completed, from pool \r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - ResultStage 5 (showString at NativeMethodAccessorImpl.java:0) finished in 0.039 s\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - Job 3 is finished. Cancelling potential speculative or zombie tasks for this job\r\n",
      "2025-01-02 13:57:43 INFO  TaskSchedulerImpl:60 - Killing all running tasks in stage 5: Stage finished\r\n",
      "2025-01-02 13:57:43 INFO  DAGScheduler:60 - Job 3 finished: showString at NativeMethodAccessorImpl.java:0, took 0.049568 s\r\n",
      "+----------+----------+\r\n",
      "|year(date)|avg(value)|\r\n",
      "+----------+----------+\r\n",
      "|      2025|       2.1|\r\n",
      "|      2020|       3.0|\r\n",
      "+----------+----------+\r\n",
      "\r\n",
      "2025-01-02 13:57:43 INFO  SparkContext:60 - SparkContext is stopping with exitCode 0.\r\n",
      "2025-01-02 13:57:43 INFO  AbstractConnector:383 - Stopped Spark@2c0b6ac4{HTTP/1.1, (http/1.1)}{0.0.0.0:4041}\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:43 INFO  SparkUI:60 - Stopped Spark web UI at http://fv-az807-413:4041\r\n",
      "2025-01-02 13:57:43 INFO  MapOutputTrackerMasterEndpoint:60 - MapOutputTrackerMasterEndpoint stopped!\r\n",
      "2025-01-02 13:57:43 INFO  MemoryStore:60 - MemoryStore cleared\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:43 INFO  BlockManager:60 - BlockManager stopped\r\n",
      "2025-01-02 13:57:43 INFO  BlockManagerMaster:60 - BlockManagerMaster stopped\r\n",
      "2025-01-02 13:57:43 INFO  OutputCommitCoordinator$OutputCommitCoordinatorEndpoint:60 - OutputCommitCoordinator stopped!\r\n",
      "2025-01-02 13:57:43 INFO  SparkContext:60 - Successfully stopped SparkContext\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:44 INFO  ShutdownHookManager:60 - Shutdown hook called\r\n",
      "2025-01-02 13:57:44 INFO  ShutdownHookManager:60 - Deleting directory /tmp/spark-39608945-24ea-4335-bb8c-f182990cbb4c/pyspark-4c7eb6f6-314e-4c6d-9991-3f17cd1a967a\r\n",
      "2025-01-02 13:57:44 INFO  ShutdownHookManager:60 - Deleting directory /tmp/spark-85869321-eecb-4187-b9e5-dc28306a1cc4\r\n",
      "2025-01-02 13:57:44 INFO  ShutdownHookManager:60 - Deleting directory /tmp/spark-39608945-24ea-4335-bb8c-f182990cbb4c\r\n"
     ]
    }
   ],
   "source": [
    "!spark-submit my_app.py"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "EBrWpshwfbzz",
   "metadata": {
    "id": "EBrWpshwfbzz"
   },
   "source": [
    "# Add your own logging messages\n",
    "\n",
    "PySpark's logging system is based on the Log4j logger and is configured in the `log4j2.properties` file.\n",
    "\n",
    "You can set up your own logging system and integrate it with PySpark's logging.\n",
    "\n",
    "We are going to showcase two scenarios:\n",
    "\n",
    "* two distinct logging systems (PySpark's logging system and Python's `logging`)\n",
    "* unified logging by redirecting PySpark's logs to Python's `logging` module"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1xaN2Gdr_FAY",
   "metadata": {
    "id": "1xaN2Gdr_FAY"
   },
   "source": [
    "## Two logging systems"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "NY13iIwMZhs7",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:57:44.355472Z",
     "iopub.status.busy": "2025-01-02T13:57:44.355245Z",
     "iopub.status.idle": "2025-01-02T13:57:44.360078Z",
     "shell.execute_reply": "2025-01-02T13:57:44.359533Z"
    },
    "id": "NY13iIwMZhs7",
    "outputId": "61249683-8d75-4caa-edc2-0480bf133cb3"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing test.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile test.py\n",
    "from pyspark.sql import SparkSession\n",
    "import logging\n",
    "\n",
    "# Create a SparkSession\n",
    "spark = SparkSession.builder \\\n",
    "   .appName(\"Logging Demo: two systems\") \\\n",
    "   .getOrCreate()\n",
    "\n",
    "# create my logger and set log level to WARN\n",
    "logging.basicConfig(level=logging.WARN)\n",
    "logger = logging.getLogger(__name__)\n",
    "\n",
    "# set PySpark log level to WARN\n",
    "sc = spark.sparkContext\n",
    "sc.setLogLevel(\"WARN\")\n",
    "\n",
    "rdd = spark.sparkContext.parallelize(range(10**5))\n",
    "logger.error(\"Computed sum: %s\", rdd.sum())\n",
    "\n",
    "# Stop the SparkSession\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "iOl4D6sodF0j",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:57:44.362114Z",
     "iopub.status.busy": "2025-01-02T13:57:44.361915Z",
     "iopub.status.idle": "2025-01-02T13:57:50.302330Z",
     "shell.execute_reply": "2025-01-02T13:57:50.301433Z"
    },
    "id": "iOl4D6sodF0j",
    "outputId": "b823bc47-ee22-4548-cabf-9bad96248ccc"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:46 WARN  NativeCodeLoader:60 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:47 WARN  Utils:72 - Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ERROR:__main__:Computed sum: 4999950000\r\n"
     ]
    }
   ],
   "source": [
    "!spark-submit test.py"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "mazhs18PH30y",
   "metadata": {
    "id": "mazhs18PH30y"
   },
   "source": [
    "## Unified logging"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "anjRU3xffgHC",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:57:50.305347Z",
     "iopub.status.busy": "2025-01-02T13:57:50.304826Z",
     "iopub.status.idle": "2025-01-02T13:57:50.309671Z",
     "shell.execute_reply": "2025-01-02T13:57:50.309025Z"
    },
    "id": "anjRU3xffgHC",
    "outputId": "dc45a0ff-2489-403e-daa6-75ee133c64dc"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting test.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile test.py\n",
    "from pyspark.sql import SparkSession\n",
    "import logging\n",
    "\n",
    "# Create a SparkSession\n",
    "spark = SparkSession.builder \\\n",
    "   .appName(\"Logging Demo: unified log system\") \\\n",
    "   .getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext\n",
    "\n",
    "# use PySpark's logger and set log level to WARN\n",
    "log4jLogger = sc._jvm.org.apache.log4j\n",
    "logger = log4jLogger.LogManager.getLogger(__name__)\n",
    "logger.setLevel(log4jLogger.Level.WARN)  # Set desired logging level\n",
    "\n",
    "# set PySpark log level to WARN\n",
    "sc.setLogLevel(\"WARN\")\n",
    "\n",
    "rdd = spark.sparkContext.parallelize(range(10**5))\n",
    "logger.warn(f\"Computed sum: {rdd.sum()}\")\n",
    "\n",
    "# Stop the SparkSession\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "NbsJ7vWXgBBg",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "execution": {
     "iopub.execute_input": "2025-01-02T13:57:50.311680Z",
     "iopub.status.busy": "2025-01-02T13:57:50.311322Z",
     "iopub.status.idle": "2025-01-02T13:57:56.113615Z",
     "shell.execute_reply": "2025-01-02T13:57:56.112732Z"
    },
    "id": "NbsJ7vWXgBBg",
    "outputId": "fdd75912-75d3-4c41-a936-9e813da82a72"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:52 WARN  NativeCodeLoader:60 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:53 WARN  Utils:72 - Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2025-01-02 13:57:54 WARN  __main__:244 - Computed sum: 4999950000\r\n"
     ]
    }
   ],
   "source": [
    "!spark-submit test.py"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "GhyGHFyVHTBx",
   "metadata": {
    "id": "GhyGHFyVHTBx"
   },
   "source": [
    "Now your warning generated by the line\n",
    "\n",
    "    logger.warn(f\"Computed sum: {rdd.sum()}\")\n",
    "\n",
    "is integrated with the Log4j messages and has the same format."
   ]
  }
 ],
 "metadata": {
  "colab": {
   "include_colab_link": true,
   "provenance": [],
   "toc_visible": true
  },
  "kernelspec": {
   "display_name": "Python 3",
   "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.8.18"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}