{ "cells": [ { "cell_type": "markdown", "metadata": { "id": "view-in-github", "colab_type": "text" }, "source": [ "\"Open" ] }, { "cell_type": "markdown", "id": "d6244506", "metadata": { "id": "d6244506" }, "source": [ "
\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", "source": [ "## Table of contents\n", "\n", "
" ], "metadata": { "id": "D5YfPFHcjDax" }, "id": "D5YfPFHcjDax" }, { "cell_type": "markdown", "source": [ "# Preliminaries\n", "\n", "The libraries needed to run this notebook. Execute this cell before any other." ], "metadata": { "id": "GGT0FAoCuefo" }, "id": "GGT0FAoCuefo" }, { "cell_type": "code", "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" ], "metadata": { "id": "MTahPuF4uetz" }, "id": "MTahPuF4uetz", "execution_count": 1, "outputs": [] }, { "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": { "id": "0fd4ce7b" }, "outputs": [], "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 }, "id": "7d8b99e3", "outputId": "e2a92282-ef3a-4f3a-ebff-f847ca66e010" }, "outputs": [ { "output_type": "execute_result", "data": { "text/plain": [ "" ], "text/html": [ "\n", "
\n", "

SparkContext

\n", "\n", "

Spark UI

\n", "\n", "
\n", "
Version
\n", "
v3.5.3
\n", "
Master
\n", "
local[*]
\n", "
AppName
\n", "
My Spark App 🌟
\n", "
\n", "
\n", " " ] }, "metadata": {}, "execution_count": 3 } ], "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 }, "id": "bed1685a", "outputId": "18f843d7-0e76-4825-ce2d-250dd5ed1182" }, "outputs": [ { "output_type": "execute_result", "data": { "text/plain": [ "'local-1735823266365'" ], "application/vnd.google.colaboratory.intrinsic+json": { "type": "string" } }, "metadata": {}, "execution_count": 4 } ], "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 }, "id": "f0023074", "outputId": "3b2bfc58-064d-4e41-b9ca-0bc069187bea" }, "outputs": [ { "output_type": "execute_result", "data": { "text/plain": [ "'local-1735823266365'" ], "application/vnd.google.colaboratory.intrinsic+json": { "type": "string" } }, "metadata": {}, "execution_count": 5 } ], "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", "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", "```" ], "metadata": { "id": "CdNLf1NpsvHn" }, "id": "CdNLf1NpsvHn" }, { "cell_type": "markdown", "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." ], "metadata": { "id": "NMfLSPZXq8Ox" }, "id": "NMfLSPZXq8Ox" }, { "cell_type": "code", "source": [ "!timeout 20 pyspark" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "N2t0ZTAvrpeR", "outputId": "910d7219-7238-4af9-8320-15cc536aabee" }, "id": "N2t0ZTAvrpeR", "execution_count": 6, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "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/02 13:07:56 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n", "25/01/02 13:07:59 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://82b3b0c373a8:4041\n", "Spark context available as 'sc' (master = local[*], app id = local-1735823279619).\n", "SparkSession available as 'spark'.\n", ">>> sc.applicationId\n", "'local-1735823279619'\n", ">>> " ] } ] }, { "cell_type": "markdown", "source": [ "Close the Spark session with `stop()`." ], "metadata": { "id": "UqaM4j78tTlG" }, "id": "UqaM4j78tTlG" }, { "cell_type": "code", "source": [ "spark.stop()" ], "metadata": { "id": "PoVb-kEdtTvj" }, "id": "PoVb-kEdtTvj", "execution_count": 7, "outputs": [] }, { "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", "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))" ], "metadata": { "id": "jmH1o6CpRR0Y" }, "id": "jmH1o6CpRR0Y" }, { "cell_type": "markdown", "source": [ "## Get or set default parallelism" ], "metadata": { "id": "pimhDh1nk4Q8" }, "id": "pimhDh1nk4Q8" }, { "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": { "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/" }, "id": "f221e3d1", "outputId": "270bf66b-d3b2-4a8e-ca09-8925823ecd37" }, "outputs": [ { "output_type": "execute_result", "data": { "text/plain": [ "2" ] }, "metadata": {}, "execution_count": 9 } ], "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": { "id": "4cf47f42" }, "outputs": [], "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/" }, "id": "6b592d7e", "outputId": "5804fc3b-24d5-4aa7-d7ad-bb68a2dcace5" }, "outputs": [ { "output_type": "execute_result", "data": { "text/plain": [ "2" ] }, "metadata": {}, "execution_count": 11 } ], "source": [ "spark.sparkContext.defaultParallelism" ] }, { "cell_type": "markdown", "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):" ], "metadata": { "id": "g3r7LB2yHDiW" }, "id": "g3r7LB2yHDiW" }, { "cell_type": "code", "source": [ "spark.conf.set(\"spark.default.parallelism\", 3)\n", "spark.sparkContext.defaultParallelism" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "pfBzJnnzGm5Q", "outputId": "91443e08-118e-4849-b793-0f1905a54c5e" }, "id": "pfBzJnnzGm5Q", "execution_count": 12, "outputs": [ { "output_type": "execute_result", "data": { "text/plain": [ "2" ] }, "metadata": {}, "execution_count": 12 } ] }, { "cell_type": "markdown", "id": "49e168d0", "metadata": { "id": "49e168d0" }, "source": [ "Stop and start session anew." ] }, { "cell_type": "code", "execution_count": 13, "id": "da0512cd", "metadata": { "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/" }, "id": "bba24e5e", "outputId": "63f25982-bc1d-4d8a-df83-7b716a6b9972" }, "outputs": [ { "output_type": "execute_result", "data": { "text/plain": [ "4" ] }, "metadata": {}, "execution_count": 14 } ], "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 }, "id": "0e2fbc63", "outputId": "18a059dc-a8b9-41d6-d775-536f3d0e03f5" }, "outputs": [ { "output_type": "execute_result", "data": { "text/plain": [ "" ], "text/html": [ "\n", "
\n", "

SparkContext

\n", "\n", "

Spark UI

\n", "\n", "
\n", "
Version
\n", "
v3.5.3
\n", "
Master
\n", "
local[*]
\n", "
AppName
\n", "
Default Parallelism 🧵🧵🧵🧵
\n", "
\n", "
\n", " " ] }, "metadata": {}, "execution_count": 15 } ], "source": [ "spark.sparkContext" ] }, { "cell_type": "markdown", "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)" ], "metadata": { "id": "2e5p9uyfFfx0" }, "id": "2e5p9uyfFfx0" }, { "cell_type": "markdown", "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." ], "metadata": { "id": "pnIik2A5lD8b" }, "id": "pnIik2A5lD8b" }, { "cell_type": "code", "source": [ "print(f\"Number of cores: {os.cpu_count()}\")" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "y_IA5tcD0wql", "outputId": "82e77377-745f-4ce8-a0c6-bba2ea6c8f42" }, "id": "y_IA5tcD0wql", "execution_count": 16, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "Number of cores: 2\n" ] } ] }, { "cell_type": "markdown", "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." ], "metadata": { "id": "sl6PQsPgvUr-" }, "id": "sl6PQsPgvUr-" }, { "cell_type": "code", "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())" ], "metadata": { "id": "1Udn9tjGvU7J" }, "id": "1Udn9tjGvU7J", "execution_count": 17, "outputs": [] }, { "cell_type": "markdown", "source": [ "Define an aggregation function that computes the sum of all numbers." ], "metadata": { "id": "u5iuCUMPD2Jv" }, "id": "u5iuCUMPD2Jv" }, { "cell_type": "code", "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" ], "metadata": { "id": "2HZ-1ENfD1Ax" }, "id": "2HZ-1ENfD1Ax", "execution_count": 18, "outputs": [] }, { "cell_type": "code", "source": [ "aggregate_data(df).collect()" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "nTXukfqEzEiA", "outputId": "d2383d0d-d2fe-4c47-9557-1cc41e9f47b7" }, "id": "nTXukfqEzEiA", "execution_count": 19, "outputs": [ { "output_type": "execute_result", "data": { "text/plain": [ "[Row(total_value=5024.487977822859, average_value=0.5024487977822859, std_deviation=0.28935206479270414, median=0.5063251524763812)]" ] }, "metadata": {}, "execution_count": 19 } ] }, { "cell_type": "markdown", "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$." ], "metadata": { "id": "RVQkbt6_1L8z" }, "id": "RVQkbt6_1L8z" }, { "cell_type": "code", "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(\"\")" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "OjUFcn0PDpb5", "outputId": "e1effc38-0c82-49bc-9baa-00195694cfa9" }, "id": "OjUFcn0PDpb5", "execution_count": 20, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "Parallelism: 1\n", "CPU times: user 351 ms, sys: 58.7 ms, total: 410 ms\n", "Wall time: 1min 9s\n", "\n", "Parallelism: 2\n", "CPU times: user 333 ms, sys: 45.7 ms, total: 379 ms\n", "Wall time: 1min 6s\n", "\n", "Parallelism: 4\n", "CPU times: user 322 ms, sys: 42.9 ms, total: 365 ms\n", "Wall time: 1min 3s\n", "\n", "Parallelism: 8\n", "CPU times: user 321 ms, sys: 41 ms, total: 362 ms\n", "Wall time: 1min 3s\n", "\n" ] } ] }, { "cell_type": "markdown", "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", "```" ], "metadata": { "id": "F_di5wA-QHdn" }, "id": "F_di5wA-QHdn" }, { "cell_type": "markdown", "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" ], "metadata": { "id": "ho2KnZNrxyea" }, "id": "ho2KnZNrxyea" }, { "cell_type": "markdown", "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!" ], "metadata": { "id": "xwXJJLUZx6Tq" }, "id": "xwXJJLUZx6Tq" }, { "cell_type": "code", "source": [ "!timeout 240 pyspark" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "lnIAoX_1TG_k", "outputId": "a56bf0e8-a500-434f-9996-6deab2d08a93" }, "id": "lnIAoX_1TG_k", "execution_count": 21, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "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/02 13:12:52 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n", "25/01/02 13:12:54 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://82b3b0c373a8:4041\n", "Spark context available as 'sc' (master = local[*], app id = local-1735823574746).\n", "SparkSession available as 'spark'.\n", ">>> from pyspark.sql.functions import year\n", ">>> from datetime import timedelta, date\n", ">>> df = spark.createDataFrame([ {\"date\": date.today(), \"value\": 2.1}, {\"date\": date.today() + timedelta(days=1), \"value\": 1.9}, {\"date\": date.today() + timedelta(days=2), \"value\": 2.3}, {\"date\": date.today() - timedelta(days=365*5), \"value\": 3.0}, ])\n", ">>> sc = spark.sparkContext\n", ">>> sc.setLogLevel(\"INFO\")\n", ">>> df.groupBy(year(\"date\")).avg().show()\n", "25/01/02 13:13:51 INFO CodeGenerator: Code generated in 517.554336 ms\n", "25/01/02 13:13:51 INFO DAGScheduler: Registering RDD 6 (showString at NativeMethodAccessorImpl.java:0) as input to shuffle 0\n", "25/01/02 13:13:51 INFO DAGScheduler: Got map stage job 0 (showString at NativeMethodAccessorImpl.java:0) with 2 output partitions\n", "25/01/02 13:13:51 INFO DAGScheduler: Final stage: ShuffleMapStage 0 (showString at NativeMethodAccessorImpl.java:0)\n", "25/01/02 13:13:51 INFO DAGScheduler: Parents of final stage: List()\n", "25/01/02 13:13:51 INFO DAGScheduler: Missing parents: List()\n", "25/01/02 13:13:51 INFO DAGScheduler: Submitting ShuffleMapStage 0 (MapPartitionsRDD[6] at showString at NativeMethodAccessorImpl.java:0), which has no missing parents\n", "25/01/02 13:13:51 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 41.0 KiB, free 434.4 MiB)\n", "25/01/02 13:13:52 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 19.2 KiB, free 434.3 MiB)\n", "25/01/02 13:13:52 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on 82b3b0c373a8:45757 (size: 19.2 KiB, free: 434.4 MiB)\n", "25/01/02 13:13:52 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1585\n", "25/01/02 13:13:52 INFO DAGScheduler: Submitting 2 missing tasks from ShuffleMapStage 0 (MapPartitionsRDD[6] at showString at NativeMethodAccessorImpl.java:0) (first 15 tasks are for partitions Vector(0, 1))\n", "25/01/02 13:13:52 INFO TaskSchedulerImpl: Adding task set 0.0 with 2 tasks resource profile 0\n", "25/01/02 13:13:52 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0) (82b3b0c373a8, executor driver, partition 0, PROCESS_LOCAL, 9022 bytes) \n", "25/01/02 13:13:52 INFO TaskSetManager: Starting task 1.0 in stage 0.0 (TID 1) (82b3b0c373a8, executor driver, partition 1, PROCESS_LOCAL, 9022 bytes) \n", "25/01/02 13:13:52 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)\n", "25/01/02 13:13:52 INFO Executor: Running task 1.0 in stage 0.0 (TID 1)\n", "25/01/02 13:13:54 INFO CodeGenerator: Code generated in 237.646408 ms\n", "25/01/02 13:13:54 INFO CodeGenerator: Code generated in 68.606548 ms\n", "25/01/02 13:13:54 INFO CodeGenerator: Code generated in 57.833732 ms\n", "25/01/02 13:13:54 INFO CodeGenerator: Code generated in 83.168364 ms\n", "25/01/02 13:13:54 INFO CodeGenerator: Code generated in 58.097344 ms\n", "25/01/02 13:13:54 INFO PythonRunner: Times: total = 2189, boot = 1261, init = 928, finish = 0\n", "25/01/02 13:13:55 INFO PythonRunner: Times: total = 2236, boot = 1276, init = 960, finish = 0\n", "25/01/02 13:13:55 INFO Executor: Finished task 1.0 in stage 0.0 (TID 1). 2882 bytes result sent to driver\n", "25/01/02 13:13:55 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 2882 bytes result sent to driver\n", "25/01/02 13:13:55 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 2995 ms on 82b3b0c373a8 (executor driver) (1/2)\n", "25/01/02 13:13:55 INFO TaskSetManager: Finished task 1.0 in stage 0.0 (TID 1) in 2959 ms on 82b3b0c373a8 (executor driver) (2/2)\n", "25/01/02 13:13:55 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool \n", "25/01/02 13:13:55 INFO PythonAccumulatorV2: Connected to AccumulatorServer at host: 127.0.0.1 port: 58885\n", "25/01/02 13:13:55 INFO DAGScheduler: ShuffleMapStage 0 (showString at NativeMethodAccessorImpl.java:0) finished in 3.595 s\n", "25/01/02 13:13:55 INFO DAGScheduler: looking for newly runnable stages\n", "25/01/02 13:13:55 INFO DAGScheduler: running: Set()\n", "25/01/02 13:13:55 INFO DAGScheduler: waiting: Set()\n", "25/01/02 13:13:55 INFO DAGScheduler: failed: Set()\n", "25/01/02 13:13:55 INFO ShufflePartitionsUtil: For shuffle(0), advisory target size: 67108864, actual target size 1048576, minimum partition size: 1048576\n", "25/01/02 13:13:55 INFO HashAggregateExec: spark.sql.codegen.aggregate.map.twolevel.enabled is set to true, but current version of codegened fast hashmap does not support this aggregate.\n", "25/01/02 13:13:55 INFO CodeGenerator: Code generated in 49.284983 ms\n", "25/01/02 13:13:55 INFO SparkContext: Starting job: showString at NativeMethodAccessorImpl.java:0\n", "25/01/02 13:13:55 INFO DAGScheduler: Got job 1 (showString at NativeMethodAccessorImpl.java:0) with 1 output partitions\n", "25/01/02 13:13:55 INFO DAGScheduler: Final stage: ResultStage 2 (showString at NativeMethodAccessorImpl.java:0)\n", "25/01/02 13:13:55 INFO DAGScheduler: Parents of final stage: List(ShuffleMapStage 1)\n", "25/01/02 13:13:55 INFO DAGScheduler: Missing parents: List()\n", "25/01/02 13:13:55 INFO DAGScheduler: Submitting ResultStage 2 (MapPartitionsRDD[9] at showString at NativeMethodAccessorImpl.java:0), which has no missing parents\n", "25/01/02 13:13:55 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 45.0 KiB, free 434.3 MiB)\n", "25/01/02 13:13:55 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 21.2 KiB, free 434.3 MiB)\n", "25/01/02 13:13:55 INFO BlockManagerInfo: Added broadcast_1_piece0 in memory on 82b3b0c373a8:45757 (size: 21.2 KiB, free: 434.4 MiB)\n", "25/01/02 13:13:55 INFO SparkContext: Created broadcast 1 from broadcast at DAGScheduler.scala:1585\n", "25/01/02 13:13:55 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 2 (MapPartitionsRDD[9] at showString at NativeMethodAccessorImpl.java:0) (first 15 tasks are for partitions Vector(0))\n", "25/01/02 13:13:55 INFO TaskSchedulerImpl: Adding task set 2.0 with 1 tasks resource profile 0\n", "25/01/02 13:13:55 INFO TaskSetManager: Starting task 0.0 in stage 2.0 (TID 2) (82b3b0c373a8, executor driver, partition 0, NODE_LOCAL, 8999 bytes) \n", "25/01/02 13:13:55 INFO Executor: Running task 0.0 in stage 2.0 (TID 2)\n", "25/01/02 13:13:55 INFO ShuffleBlockFetcherIterator: Getting 2 (216.0 B) non-empty blocks including 2 (216.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\n", "25/01/02 13:13:55 INFO ShuffleBlockFetcherIterator: Started 0 remote fetches in 16 ms\n", "25/01/02 13:13:55 INFO CodeGenerator: Code generated in 29.145021 ms\n", "25/01/02 13:13:55 INFO Executor: Finished task 0.0 in stage 2.0 (TID 2). 5316 bytes result sent to driver\n", "25/01/02 13:13:55 INFO TaskSetManager: Finished task 0.0 in stage 2.0 (TID 2) in 193 ms on 82b3b0c373a8 (executor driver) (1/1)\n", "25/01/02 13:13:55 INFO DAGScheduler: ResultStage 2 (showString at NativeMethodAccessorImpl.java:0) finished in 0.219 s\n", "25/01/02 13:13:55 INFO DAGScheduler: Job 1 is finished. Cancelling potential speculative or zombie tasks for this job\n", "25/01/02 13:13:55 INFO TaskSchedulerImpl: Removed TaskSet 2.0, whose tasks have all completed, from pool \n", "25/01/02 13:13:55 INFO TaskSchedulerImpl: Killing all running tasks in stage 2: Stage finished\n", "25/01/02 13:13:55 INFO DAGScheduler: Job 1 finished: showString at NativeMethodAccessorImpl.java:0, took 0.253734 s\n", "25/01/02 13:13:57 INFO CodeGenerator: Code generated in 12.529908 ms\n", "+----------+----------+\n", "|year(date)|avg(value)|\n", "+----------+----------+\n", "| 2025| 2.1|\n", "| 2020| 3.0|\n", "+----------+----------+\n", "\n", ">>> sc.setLogLevel(\"WARN\")\n", ">>> df.groupBy(year(\"date\")).avg().show()\n", "+----------+----------+\n", "|year(date)|avg(value)|\n", "+----------+----------+\n", "| 2025| 2.1|\n", "| 2020| 3.0|\n", "+----------+----------+\n", "\n", ">>> quit()\n" ] } ] }, { "cell_type": "markdown", "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", "```" ], "metadata": { "id": "kIsl8s5jWG5S" }, "id": "kIsl8s5jWG5S" }, { "cell_type": "markdown", "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)." ], "metadata": { "id": "r2VALwrWV5Tw" }, "id": "r2VALwrWV5Tw" }, { "cell_type": "code", "source": [ "!find_spark_home.py" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "2cppX3R-UdvB", "outputId": "722fa43c-36e0-4ca5-c983-75c83ab68354" }, "id": "2cppX3R-UdvB", "execution_count": 22, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "/usr/local/lib/python3.10/dist-packages/pyspark\n" ] } ] }, { "cell_type": "code", "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()" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "BUhX4x1tWxtC", "outputId": "caa13b29-5511-4f86-f1c8-0026a1efb983" }, "id": "BUhX4x1tWxtC", "execution_count": 23, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "Output of find_spark_home.py: /usr/local/lib/python3.10/dist-packages/pyspark\n", "\n" ] } ] }, { "cell_type": "markdown", "source": [ "Now the variable `SPARK_HOME` is set." ], "metadata": { "id": "FAI_9pw1qrbJ" }, "id": "FAI_9pw1qrbJ" }, { "cell_type": "code", "source": [ "!echo $SPARK_HOME" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "sQHZdV5EW_DL", "outputId": "1e02f290-2d99-46a4-a8c6-95c109912923" }, "id": "sQHZdV5EW_DL", "execution_count": 24, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "/usr/local/lib/python3.10/dist-packages/pyspark\n" ] } ] }, { "cell_type": "markdown", "source": [ "Create a `log4j2.properties` file in Spark's configuration directory." ], "metadata": { "id": "HH7_bAv6qe14" }, "id": "HH7_bAv6qe14" }, { "cell_type": "code", "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", "🤖" ], "metadata": { "id": "W96FZEFcWcPo" }, "id": "W96FZEFcWcPo", "execution_count": 25, "outputs": [] }, { "cell_type": "code", "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()" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "gmKvfBl3XkzO", "outputId": "d941f999-3819-41d7-9efa-ca50d447423b" }, "id": "gmKvfBl3XkzO", "execution_count": 26, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "Writing my_app.py\n" ] } ] }, { "cell_type": "code", "source": [ "!spark-submit my_app.py" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "qwNPA_BN5sQs", "outputId": "cf91872b-0eb7-40d5-8173-cd126680f1ea" }, "id": "qwNPA_BN5sQs", "execution_count": 27, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "INFO:__main__:Creating Spark session\n", "2025-01-02 13:14:24 WARN NativeCodeLoader:60 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n", "2025-01-02 13:14:25 WARN Utils:72 - Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n", "INFO:numexpr.utils:NumExpr defaulting to 2 threads.\n", "+----------+----------+\n", "|year(date)|avg(value)|\n", "+----------+----------+\n", "| 2025| 2.1|\n", "| 2020| 3.0|\n", "+----------+----------+\n", "\n", "ERROR:__main__:Setting log level to INFO\n", "2025-01-02 13:14:38 INFO DAGScheduler:60 - Registering RDD 11 (showString at NativeMethodAccessorImpl.java:0) as input to shuffle 1\n", "2025-01-02 13:14:38 INFO DAGScheduler:60 - Got map stage job 2 (showString at NativeMethodAccessorImpl.java:0) with 2 output partitions\n", "2025-01-02 13:14:38 INFO DAGScheduler:60 - Final stage: ShuffleMapStage 3 (showString at NativeMethodAccessorImpl.java:0)\n", "2025-01-02 13:14:38 INFO DAGScheduler:60 - Parents of final stage: List()\n", "2025-01-02 13:14:38 INFO DAGScheduler:60 - Missing parents: List()\n", "2025-01-02 13:14:38 INFO DAGScheduler:60 - Submitting ShuffleMapStage 3 (MapPartitionsRDD[11] at showString at NativeMethodAccessorImpl.java:0), which has no missing parents\n", "2025-01-02 13:14:38 INFO MemoryStore:60 - Block broadcast_2 stored as values in memory (estimated size 40.9 KiB, free 434.4 MiB)\n", "2025-01-02 13:14:38 INFO MemoryStore:60 - Block broadcast_2_piece0 stored as bytes in memory (estimated size 19.2 KiB, free 434.3 MiB)\n", "2025-01-02 13:14:38 INFO BlockManagerInfo:60 - Added broadcast_2_piece0 in memory on 82b3b0c373a8:44451 (size: 19.2 KiB, free: 434.4 MiB)\n", "2025-01-02 13:14:38 INFO SparkContext:60 - Created broadcast 2 from broadcast at DAGScheduler.scala:1585\n", "2025-01-02 13:14:38 INFO DAGScheduler:60 - Submitting 2 missing tasks from ShuffleMapStage 3 (MapPartitionsRDD[11] at showString at NativeMethodAccessorImpl.java:0) (first 15 tasks are for partitions Vector(0, 1))\n", "2025-01-02 13:14:38 INFO TaskSchedulerImpl:60 - Adding task set 3.0 with 2 tasks resource profile 0\n", "2025-01-02 13:14:38 INFO TaskSetManager:60 - Starting task 0.0 in stage 3.0 (TID 3) (82b3b0c373a8, executor driver, partition 0, PROCESS_LOCAL, 9022 bytes) \n", "2025-01-02 13:14:38 INFO TaskSetManager:60 - Starting task 1.0 in stage 3.0 (TID 4) (82b3b0c373a8, executor driver, partition 1, PROCESS_LOCAL, 9022 bytes) \n", "2025-01-02 13:14:38 INFO Executor:60 - Running task 0.0 in stage 3.0 (TID 3)\n", "2025-01-02 13:14:38 INFO Executor:60 - Running task 1.0 in stage 3.0 (TID 4)\n", "2025-01-02 13:14:39 INFO PythonRunner:60 - Times: total = 265, boot = -3034, init = 3298, finish = 1\n", "2025-01-02 13:14:39 INFO PythonRunner:60 - Times: total = 291, boot = -2895, init = 3186, finish = 0\n", "2025-01-02 13:14:39 INFO Executor:60 - Finished task 0.0 in stage 3.0 (TID 3). 2839 bytes result sent to driver\n", "2025-01-02 13:14:39 INFO TaskSetManager:60 - Finished task 0.0 in stage 3.0 (TID 3) in 352 ms on 82b3b0c373a8 (executor driver) (1/2)\n", "2025-01-02 13:14:39 INFO Executor:60 - Finished task 1.0 in stage 3.0 (TID 4). 2839 bytes result sent to driver\n", "2025-01-02 13:14:39 INFO TaskSetManager:60 - Finished task 1.0 in stage 3.0 (TID 4) in 355 ms on 82b3b0c373a8 (executor driver) (2/2)\n", "2025-01-02 13:14:39 INFO TaskSchedulerImpl:60 - Removed TaskSet 3.0, whose tasks have all completed, from pool \n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - ShuffleMapStage 3 (showString at NativeMethodAccessorImpl.java:0) finished in 0.374 s\n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - looking for newly runnable stages\n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - running: Set()\n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - waiting: Set()\n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - failed: Set()\n", "2025-01-02 13:14:39 INFO ShufflePartitionsUtil:60 - For shuffle(1), advisory target size: 67108864, actual target size 1048576, minimum partition size: 1048576\n", "2025-01-02 13:14:39 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.\n", "2025-01-02 13:14:39 INFO SparkContext:60 - Starting job: showString at NativeMethodAccessorImpl.java:0\n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - Got job 3 (showString at NativeMethodAccessorImpl.java:0) with 1 output partitions\n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - Final stage: ResultStage 5 (showString at NativeMethodAccessorImpl.java:0)\n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - Parents of final stage: List(ShuffleMapStage 4)\n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - Missing parents: List()\n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - Submitting ResultStage 5 (MapPartitionsRDD[14] at showString at NativeMethodAccessorImpl.java:0), which has no missing parents\n", "2025-01-02 13:14:39 INFO MemoryStore:60 - Block broadcast_3 stored as values in memory (estimated size 45.0 KiB, free 434.3 MiB)\n", "2025-01-02 13:14:39 INFO MemoryStore:60 - Block broadcast_3_piece0 stored as bytes in memory (estimated size 21.2 KiB, free 434.3 MiB)\n", "2025-01-02 13:14:39 INFO BlockManagerInfo:60 - Added broadcast_3_piece0 in memory on 82b3b0c373a8:44451 (size: 21.2 KiB, free: 434.4 MiB)\n", "2025-01-02 13:14:39 INFO SparkContext:60 - Created broadcast 3 from broadcast at DAGScheduler.scala:1585\n", "2025-01-02 13:14:39 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))\n", "2025-01-02 13:14:39 INFO TaskSchedulerImpl:60 - Adding task set 5.0 with 1 tasks resource profile 0\n", "2025-01-02 13:14:39 INFO TaskSetManager:60 - Starting task 0.0 in stage 5.0 (TID 5) (82b3b0c373a8, executor driver, partition 0, NODE_LOCAL, 8999 bytes) \n", "2025-01-02 13:14:39 INFO Executor:60 - Running task 0.0 in stage 5.0 (TID 5)\n", "2025-01-02 13:14:39 INFO ShuffleBlockFetcherIterator:60 - Getting 2 (216.0 B) non-empty blocks including 2 (216.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\n", "2025-01-02 13:14:39 INFO ShuffleBlockFetcherIterator:60 - Started 0 remote fetches in 4 ms\n", "2025-01-02 13:14:39 INFO Executor:60 - Finished task 0.0 in stage 5.0 (TID 5). 5273 bytes result sent to driver\n", "2025-01-02 13:14:39 INFO TaskSetManager:60 - Finished task 0.0 in stage 5.0 (TID 5) in 52 ms on 82b3b0c373a8 (executor driver) (1/1)\n", "2025-01-02 13:14:39 INFO TaskSchedulerImpl:60 - Removed TaskSet 5.0, whose tasks have all completed, from pool \n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - ResultStage 5 (showString at NativeMethodAccessorImpl.java:0) finished in 0.077 s\n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - Job 3 is finished. Cancelling potential speculative or zombie tasks for this job\n", "2025-01-02 13:14:39 INFO TaskSchedulerImpl:60 - Killing all running tasks in stage 5: Stage finished\n", "2025-01-02 13:14:39 INFO DAGScheduler:60 - Job 3 finished: showString at NativeMethodAccessorImpl.java:0, took 0.100031 s\n", "+----------+----------+\n", "|year(date)|avg(value)|\n", "+----------+----------+\n", "| 2025| 2.1|\n", "| 2020| 3.0|\n", "+----------+----------+\n", "\n", "2025-01-02 13:14:39 INFO SparkContext:60 - SparkContext is stopping with exitCode 0.\n", "2025-01-02 13:14:39 INFO AbstractConnector:383 - Stopped Spark@70fe4335{HTTP/1.1, (http/1.1)}{0.0.0.0:4041}\n", "2025-01-02 13:14:39 INFO SparkUI:60 - Stopped Spark web UI at http://82b3b0c373a8:4041\n", "2025-01-02 13:14:39 INFO MapOutputTrackerMasterEndpoint:60 - MapOutputTrackerMasterEndpoint stopped!\n", "2025-01-02 13:14:39 INFO MemoryStore:60 - MemoryStore cleared\n", "2025-01-02 13:14:39 INFO BlockManager:60 - BlockManager stopped\n", "2025-01-02 13:14:39 INFO BlockManagerMaster:60 - BlockManagerMaster stopped\n", "2025-01-02 13:14:39 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint:60 - OutputCommitCoordinator stopped!\n", "2025-01-02 13:14:39 INFO SparkContext:60 - Successfully stopped SparkContext\n", "INFO:py4j.clientserver:Closing down clientserver connection\n", "2025-01-02 13:14:40 INFO ShutdownHookManager:60 - Shutdown hook called\n", "2025-01-02 13:14:40 INFO ShutdownHookManager:60 - Deleting directory /tmp/spark-dafff29b-fd92-4d4e-a958-7575b0ae52a6\n", "2025-01-02 13:14:40 INFO ShutdownHookManager:60 - Deleting directory /tmp/spark-dafff29b-fd92-4d4e-a958-7575b0ae52a6/pyspark-7626ba41-9b34-4a2b-abcf-67892e474185\n", "2025-01-02 13:14:40 INFO ShutdownHookManager:60 - Deleting directory /tmp/spark-ea02e098-5cac-4055-b258-09f66acc97d1\n" ] } ] }, { "cell_type": "markdown", "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" ], "metadata": { "id": "EBrWpshwfbzz" }, "id": "EBrWpshwfbzz" }, { "cell_type": "markdown", "source": [ "## Two logging systems" ], "metadata": { "id": "1xaN2Gdr_FAY" }, "id": "1xaN2Gdr_FAY" }, { "cell_type": "code", "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()" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "NY13iIwMZhs7", "outputId": "61249683-8d75-4caa-edc2-0480bf133cb3" }, "id": "NY13iIwMZhs7", "execution_count": 28, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "Writing test.py\n" ] } ] }, { "cell_type": "code", "source": [ "!spark-submit test.py" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "iOl4D6sodF0j", "outputId": "b823bc47-ee22-4548-cabf-9bad96248ccc" }, "id": "iOl4D6sodF0j", "execution_count": 29, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "2025-01-02 13:14:45 WARN NativeCodeLoader:60 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n", "2025-01-02 13:14:47 WARN Utils:72 - Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n", "ERROR:__main__:Computed sum: 4999950000\n" ] } ] }, { "cell_type": "markdown", "source": [ "## Unified logging" ], "metadata": { "id": "mazhs18PH30y" }, "id": "mazhs18PH30y" }, { "cell_type": "code", "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()" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "anjRU3xffgHC", "outputId": "dc45a0ff-2489-403e-daa6-75ee133c64dc" }, "id": "anjRU3xffgHC", "execution_count": 30, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "Overwriting test.py\n" ] } ] }, { "cell_type": "code", "source": [ "!spark-submit test.py" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "NbsJ7vWXgBBg", "outputId": "fdd75912-75d3-4c41-a936-9e813da82a72" }, "id": "NbsJ7vWXgBBg", "execution_count": 31, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "2025-01-02 13:14:57 WARN NativeCodeLoader:60 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n", "2025-01-02 13:14:59 WARN Utils:72 - Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n", "2025-01-02 13:15:03 WARN __main__:244 - Computed sum: 4999950000\n" ] } ] }, { "cell_type": "markdown", "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": { "id": "GhyGHFyVHTBx" }, "id": "GhyGHFyVHTBx" } ], "metadata": { "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.9.7" }, "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 }, "colab": { "provenance": [], "toc_visible": true, "include_colab_link": true } }, "nbformat": 4, "nbformat_minor": 5 }