Notebook

PySpark miscellanea¶

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.

(from: https://spark.apache.org/docs/latest/api/python/index.html)

In this notebook, we showcase various tips, tricks, and insights related to PySpark.

Preliminaries¶

The libraries needed to run this notebook. Execute this cell before any other.

In [1]:

from pyspark.sql import SparkSession
from pyspark.sql.functions import rand, sum, avg, stddev, expr, year
from datetime import timedelta, date
import os
import subprocess

How to get your application's ID¶

From the Spark session¶

What is a Spark session?

In [2]:

spark = SparkSession \
    .builder \
    .appName("My Spark App 🌟") \
    .getOrCreate()

Get the session's context (what is a Spark context? and detailed documentation).

In [3]:

sc = spark.sparkContext
sc

Out[3]:

SparkContext

Spark UI

Version: v3.5.3
Master: local[*]
AppName: My Spark App 🌟

Get applicationId from the context sc.

In [4]:

sc.applicationId

Out[4]:

'local-1735823266365'

Or in one single step:

In [5]:

spark.sparkContext.applicationId

Out[5]:

'local-1735823266365'

In the PySpark shell¶

If you're using the PySpark shell (see using the shell), SparkContext is created automatically and it can be accessed from the variable called sc.

Python 3.10.12 (main, Nov  6 2024, 20:22:13) [GCC 11.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
25/01/01 11:40:51 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
25/01/01 11:40:52 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 3.5.3
      /_/

Using Python version 3.10.12 (main, Nov  6 2024 20:22:13)
Spark context Web UI available at http://2c6bcae43959:4041
Spark context available as 'sc' (master = local[*], app id = local-1735731652766).
SparkSession available as 'spark'.
>>> sc.applicationId
'local-1735731652766'
>>> quit()

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).

We are using the timeout function (credit: https://stackoverflow.com/a/52975118) to prevent the notebook from getting stuck when being executed automatically.

In [6]:

!timeout 20 pyspark

Python 3.10.12 (main, Nov  6 2024, 20:22:13) [GCC 11.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
25/01/02 13:07:56 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
25/01/02 13:07:59 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 3.5.3
      /_/

Using Python version 3.10.12 (main, Nov  6 2024 20:22:13)
Spark context Web UI available at http://82b3b0c373a8:4041
Spark context available as 'sc' (master = local[*], app id = local-1735823279619).
SparkSession available as 'spark'.
>>> sc.applicationId
'local-1735823279619'
>>>

Close the Spark session with stop().

In [7]:

spark.stop()

Default parallelism¶

What is `spark.default.parallelism`?¶

This property determines the default number of chunks in which an RDD (Resilient Distributed Dataset) is partitioned. This, in turn, affects how many tasks are executed concurrently.

Unless specified by the user, the default value of default.parallelism is set based on the cluster manager:

in standalone mode it is equal to the number of (virtual) cores on the local machine
in Mesos: 8
for YARN, Kubernetes: total number of cores on all executor nodes or 2, whichever is larger

(see Spark configuration/Execution behavior)

Get or set default parallelism¶

Create a Spark session.

In [8]:

spark = SparkSession \
        .builder \
        .appName("Default Parallelism 🧵🧵") \
        .getOrCreate()

Show the value of defaultParallelism:

In [9]:

spark.sparkContext.defaultParallelism

Out[9]:

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!

In [10]:

spark = SparkSession \
        .builder \
        .config("spark.default.parallelism", 4) \
        .getOrCreate()

Default parallelism hasn't changed!

In [11]:

spark.sparkContext.defaultParallelism

Out[11]:

Same with SparkSession.conf.set:

In [12]:

spark.conf.set("spark.default.parallelism", 3)
spark.sparkContext.defaultParallelism

Out[12]:

Stop and start session anew.

In [13]:

spark.stop()
spark = SparkSession \
        .builder \
        .appName("Default Parallelism 🧵🧵🧵🧵") \
        .config("spark.default.parallelism", 4) \
        .getOrCreate()

In [14]:

spark.sparkContext.defaultParallelism

Out[14]:

Great! Now the context has been changed (and also the applications's name has been updated).

In [15]:

spark.sparkContext

Out[15]:

SparkContext

Spark UI

Version: v3.5.3
Master: local[*]
AppName: Default Parallelism 🧵🧵🧵🧵

The reason why you cannot change a "running" context is that

Once a SparkConf object is passed to Spark, it is cloned and can no longer be modified by the user.

(see SparkConf in the PySpark API documentation)

Example of how parallelism can influence the runtime of computation¶

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.

The standard Google Colab notebook has two cores, so $2$ is the maximum parallelization that can be achieved.

In [16]:

print(f"Number of cores: {os.cpu_count()}")

Number of cores: 2

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.

In [17]:

# Create SparkSession
spark = SparkSession.builder.appName("Parallelism Demo ⚙️⚙️").getOrCreate()

# Create a DataFrame with random numbers
df = spark.range(10**4).withColumn("value", rand())

Define an aggregation function that computes the sum of all numbers.

In [18]:

def aggregate_data(data_frame):
    result = data_frame.groupBy().agg(
        sum("value").alias("total_value"),
        avg("value").alias("average_value"),
        stddev("value").alias("std_deviation"),
        expr("percentile_approx(value, 0.5)").alias("median")  # Approximate median
    )
    return result

In [19]:

aggregate_data(df).collect()

Out[19]:

[Row(total_value=5024.487977822859, average_value=0.5024487977822859, std_deviation=0.28935206479270414, median=0.5063251524763812)]

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!).

⚠️ 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$.

In [20]:

for parallelism in [1, 2, 4, 8]:
  # Create SparkSession
  spark.stop()
  spark = SparkSession.builder.appName(f"Default Parallelism {'🧵'*parallelism}") \
                      .config("spark.default.parallelism", parallelism) \
                      .getOrCreate()
  # Create a DataFrame with random numbers
  df = spark.range(10**8).withColumn("value", rand())

  print(f"Parallelism: {spark.sparkContext.getConf().get('spark.default.parallelism')}")
  %time aggregate_data(df).collect() # Trigger the computation
  print("")

Parallelism: 1
CPU times: user 351 ms, sys: 58.7 ms, total: 410 ms
Wall time: 1min 9s

Parallelism: 2
CPU times: user 333 ms, sys: 45.7 ms, total: 379 ms
Wall time: 1min 6s

Parallelism: 4
CPU times: user 322 ms, sys: 42.9 ms, total: 365 ms
Wall time: 1min 3s

Parallelism: 8
CPU times: user 321 ms, sys: 41 ms, total: 362 ms
Wall time: 1min 3s

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.

Parallelism: 1
CPU times: user 8.22 ms, sys: 3.96 ms, total: 12.2 ms                           
Wall time: 42.4 s

Parallelism: 2
CPU times: user 4.96 ms, sys: 1.97 ms, total: 6.94 ms                           
Wall time: 23 s

Parallelism: 4
CPU times: user 4.27 ms, sys: 1.66 ms, total: 5.93 ms                           
Wall time: 16 s

Parallelism: 8
CPU times: user 4.58 ms, sys: 1.78 ms, total: 6.36 ms                           
Wall time: 15 s

How to change PySpark's log level¶

Log levels in PySpark sorted from the most verbose to the least are:

ALL
TRACE
DEBUG
INFO
WARN
ERROR
FATAL
OFF

See https://spark.apache.org/docs/.../api/pyspark.SparkContext.setLogLevel.html.

In the PySpark shell¶

To change the log level to "INFO" in the PySpark shell just enter:

sc.setLogLevel("INFO")

⚠️ Anything log level above "INFO" is extermely verbose, so be prepared for a lot of output!

In [21]:

!timeout 240 pyspark

Python 3.10.12 (main, Nov  6 2024, 20:22:13) [GCC 11.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
25/01/02 13:12:52 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
25/01/02 13:12:54 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 3.5.3
      /_/

Using Python version 3.10.12 (main, Nov  6 2024 20:22:13)
Spark context Web UI available at http://82b3b0c373a8:4041
Spark context available as 'sc' (master = local[*], app id = local-1735823574746).
SparkSession available as 'spark'.
>>> from pyspark.sql.functions import year
>>> from datetime import timedelta, date
>>> 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}, ])
>>> sc = spark.sparkContext
>>> sc.setLogLevel("INFO")
>>> df.groupBy(year("date")).avg().show()
25/01/02 13:13:51 INFO CodeGenerator: Code generated in 517.554336 ms
25/01/02 13:13:51 INFO DAGScheduler: Registering RDD 6 (showString at NativeMethodAccessorImpl.java:0) as input to shuffle 0
25/01/02 13:13:51 INFO DAGScheduler: Got map stage job 0 (showString at NativeMethodAccessorImpl.java:0) with 2 output partitions
25/01/02 13:13:51 INFO DAGScheduler: Final stage: ShuffleMapStage 0 (showString at NativeMethodAccessorImpl.java:0)
25/01/02 13:13:51 INFO DAGScheduler: Parents of final stage: List()
25/01/02 13:13:51 INFO DAGScheduler: Missing parents: List()
25/01/02 13:13:51 INFO DAGScheduler: Submitting ShuffleMapStage 0 (MapPartitionsRDD[6] at showString at NativeMethodAccessorImpl.java:0), which has no missing parents
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)
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)
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)
25/01/02 13:13:52 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1585
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))
25/01/02 13:13:52 INFO TaskSchedulerImpl: Adding task set 0.0 with 2 tasks resource profile 0
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) 
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) 
25/01/02 13:13:52 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
25/01/02 13:13:52 INFO Executor: Running task 1.0 in stage 0.0 (TID 1)
25/01/02 13:13:54 INFO CodeGenerator: Code generated in 237.646408 ms
25/01/02 13:13:54 INFO CodeGenerator: Code generated in 68.606548 ms
25/01/02 13:13:54 INFO CodeGenerator: Code generated in 57.833732 ms
25/01/02 13:13:54 INFO CodeGenerator: Code generated in 83.168364 ms
25/01/02 13:13:54 INFO CodeGenerator: Code generated in 58.097344 ms
25/01/02 13:13:54 INFO PythonRunner: Times: total = 2189, boot = 1261, init = 928, finish = 0
25/01/02 13:13:55 INFO PythonRunner: Times: total = 2236, boot = 1276, init = 960, finish = 0
25/01/02 13:13:55 INFO Executor: Finished task 1.0 in stage 0.0 (TID 1). 2882 bytes result sent to driver
25/01/02 13:13:55 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 2882 bytes result sent to driver
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)
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)
25/01/02 13:13:55 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool 
25/01/02 13:13:55 INFO PythonAccumulatorV2: Connected to AccumulatorServer at host: 127.0.0.1 port: 58885
25/01/02 13:13:55 INFO DAGScheduler: ShuffleMapStage 0 (showString at NativeMethodAccessorImpl.java:0) finished in 3.595 s
25/01/02 13:13:55 INFO DAGScheduler: looking for newly runnable stages
25/01/02 13:13:55 INFO DAGScheduler: running: Set()
25/01/02 13:13:55 INFO DAGScheduler: waiting: Set()
25/01/02 13:13:55 INFO DAGScheduler: failed: Set()
25/01/02 13:13:55 INFO ShufflePartitionsUtil: For shuffle(0), advisory target size: 67108864, actual target size 1048576, minimum partition size: 1048576
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.
25/01/02 13:13:55 INFO CodeGenerator: Code generated in 49.284983 ms
25/01/02 13:13:55 INFO SparkContext: Starting job: showString at NativeMethodAccessorImpl.java:0
25/01/02 13:13:55 INFO DAGScheduler: Got job 1 (showString at NativeMethodAccessorImpl.java:0) with 1 output partitions
25/01/02 13:13:55 INFO DAGScheduler: Final stage: ResultStage 2 (showString at NativeMethodAccessorImpl.java:0)
25/01/02 13:13:55 INFO DAGScheduler: Parents of final stage: List(ShuffleMapStage 1)
25/01/02 13:13:55 INFO DAGScheduler: Missing parents: List()
25/01/02 13:13:55 INFO DAGScheduler: Submitting ResultStage 2 (MapPartitionsRDD[9] at showString at NativeMethodAccessorImpl.java:0), which has no missing parents
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)
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)
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)
25/01/02 13:13:55 INFO SparkContext: Created broadcast 1 from broadcast at DAGScheduler.scala:1585
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))
25/01/02 13:13:55 INFO TaskSchedulerImpl: Adding task set 2.0 with 1 tasks resource profile 0
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) 
25/01/02 13:13:55 INFO Executor: Running task 0.0 in stage 2.0 (TID 2)
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
25/01/02 13:13:55 INFO ShuffleBlockFetcherIterator: Started 0 remote fetches in 16 ms
25/01/02 13:13:55 INFO CodeGenerator: Code generated in 29.145021 ms
25/01/02 13:13:55 INFO Executor: Finished task 0.0 in stage 2.0 (TID 2). 5316 bytes result sent to driver
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)
25/01/02 13:13:55 INFO DAGScheduler: ResultStage 2 (showString at NativeMethodAccessorImpl.java:0) finished in 0.219 s
25/01/02 13:13:55 INFO DAGScheduler: Job 1 is finished. Cancelling potential speculative or zombie tasks for this job
25/01/02 13:13:55 INFO TaskSchedulerImpl: Removed TaskSet 2.0, whose tasks have all completed, from pool 
25/01/02 13:13:55 INFO TaskSchedulerImpl: Killing all running tasks in stage 2: Stage finished
25/01/02 13:13:55 INFO DAGScheduler: Job 1 finished: showString at NativeMethodAccessorImpl.java:0, took 0.253734 s
25/01/02 13:13:57 INFO CodeGenerator: Code generated in 12.529908 ms
+----------+----------+
|year(date)|avg(value)|
+----------+----------+
|      2025|       2.1|
|      2020|       3.0|
+----------+----------+

>>> sc.setLogLevel("WARN")
>>> df.groupBy(year("date")).avg().show()
+----------+----------+
|year(date)|avg(value)|
+----------+----------+
|      2025|       2.1|
|      2020|       3.0|
+----------+----------+

>>> quit()

from pyspark.sql.functions import year
from datetime import timedelta, date

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},
])

sc = spark.sparkContext
sc.setLogLevel("INFO")
df.groupBy(year("date")).avg().show()
sc.setLogLevel("WARN")
df.groupBy(year("date")).avg().show()

In a PySpark script¶

After finding $SPARK_HOME, we are going to create a Log4J configuration file and finally run a PySpark script showcasing different log levels.

See also: PySpark on Google Colab.

In [22]:

!find_spark_home.py

/usr/local/lib/python3.10/dist-packages/pyspark

In [23]:

# Run the script and capture its output
result = subprocess.run(["find_spark_home.py"], capture_output=True, text=True)

# Print or use the captured output
print("Output of find_spark_home.py:", result.stdout)

# set SPARK_HOME environment variable
os.environ['SPARK_HOME'] = result.stdout.strip()

Output of find_spark_home.py: /usr/local/lib/python3.10/dist-packages/pyspark

Now the variable SPARK_HOME is set.

In [24]:

!echo $SPARK_HOME

/usr/local/lib/python3.10/dist-packages/pyspark

Create a log4j2.properties file in Spark's configuration directory.

In [25]:

%%bash
# create conf directory
# with the option -p mkdir won't complain if the folder already exists
mkdir -p $SPARK_HOME/conf

# populate log4j2.properties file
FILE=$SPARK_HOME/conf/log4j2.properties

# read about heredocs: https://tldp.org/LDP/abs/html/here-docs.html
cat> $FILE <<🤖
status = warn

appender.console.type = Console
appender.console.name = STDOUT
appender.console.target = SYSTEM_ERR

rootLogger.level = warn
rootLogger.appenderRef.stdout.ref = STDOUT

# formatting
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n
🤖

In [26]:

%%writefile my_app.py
from pyspark.sql import SparkSession
from datetime import timedelta, date
from pyspark.sql.functions import year
import logging

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)

    logger.info("Creating Spark session")
    spark = SparkSession.builder.appName("Logging levels 📝").getOrCreate()

    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},
    ])

    df.groupBy(year("date")).avg().show()
    sc = spark.sparkContext
    logger.error("Setting log level to INFO")
    sc.setLogLevel("INFO")
    df.groupBy(year("date")).avg().show()
    spark.stop()

Writing my_app.py

In [27]:

!spark-submit my_app.py

INFO:__main__:Creating Spark session
2025-01-02 13:14:24 WARN  NativeCodeLoader:60 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2025-01-02 13:14:25 WARN  Utils:72 - Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
INFO:numexpr.utils:NumExpr defaulting to 2 threads.
+----------+----------+
|year(date)|avg(value)|
+----------+----------+
|      2025|       2.1|
|      2020|       3.0|
+----------+----------+

ERROR:__main__:Setting log level to INFO
2025-01-02 13:14:38 INFO  DAGScheduler:60 - Registering RDD 11 (showString at NativeMethodAccessorImpl.java:0) as input to shuffle 1
2025-01-02 13:14:38 INFO  DAGScheduler:60 - Got map stage job 2 (showString at NativeMethodAccessorImpl.java:0) with 2 output partitions
2025-01-02 13:14:38 INFO  DAGScheduler:60 - Final stage: ShuffleMapStage 3 (showString at NativeMethodAccessorImpl.java:0)
2025-01-02 13:14:38 INFO  DAGScheduler:60 - Parents of final stage: List()
2025-01-02 13:14:38 INFO  DAGScheduler:60 - Missing parents: List()
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
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)
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)
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)
2025-01-02 13:14:38 INFO  SparkContext:60 - Created broadcast 2 from broadcast at DAGScheduler.scala:1585
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))
2025-01-02 13:14:38 INFO  TaskSchedulerImpl:60 - Adding task set 3.0 with 2 tasks resource profile 0
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) 
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) 
2025-01-02 13:14:38 INFO  Executor:60 - Running task 0.0 in stage 3.0 (TID 3)
2025-01-02 13:14:38 INFO  Executor:60 - Running task 1.0 in stage 3.0 (TID 4)
2025-01-02 13:14:39 INFO  PythonRunner:60 - Times: total = 265, boot = -3034, init = 3298, finish = 1
2025-01-02 13:14:39 INFO  PythonRunner:60 - Times: total = 291, boot = -2895, init = 3186, finish = 0
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
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)
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
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)
2025-01-02 13:14:39 INFO  TaskSchedulerImpl:60 - Removed TaskSet 3.0, whose tasks have all completed, from pool 
2025-01-02 13:14:39 INFO  DAGScheduler:60 - ShuffleMapStage 3 (showString at NativeMethodAccessorImpl.java:0) finished in 0.374 s
2025-01-02 13:14:39 INFO  DAGScheduler:60 - looking for newly runnable stages
2025-01-02 13:14:39 INFO  DAGScheduler:60 - running: Set()
2025-01-02 13:14:39 INFO  DAGScheduler:60 - waiting: Set()
2025-01-02 13:14:39 INFO  DAGScheduler:60 - failed: Set()
2025-01-02 13:14:39 INFO  ShufflePartitionsUtil:60 - For shuffle(1), advisory target size: 67108864, actual target size 1048576, minimum partition size: 1048576
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.
2025-01-02 13:14:39 INFO  SparkContext:60 - Starting job: showString at NativeMethodAccessorImpl.java:0
2025-01-02 13:14:39 INFO  DAGScheduler:60 - Got job 3 (showString at NativeMethodAccessorImpl.java:0) with 1 output partitions
2025-01-02 13:14:39 INFO  DAGScheduler:60 - Final stage: ResultStage 5 (showString at NativeMethodAccessorImpl.java:0)
2025-01-02 13:14:39 INFO  DAGScheduler:60 - Parents of final stage: List(ShuffleMapStage 4)
2025-01-02 13:14:39 INFO  DAGScheduler:60 - Missing parents: List()
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
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)
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)
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)
2025-01-02 13:14:39 INFO  SparkContext:60 - Created broadcast 3 from broadcast at DAGScheduler.scala:1585
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))
2025-01-02 13:14:39 INFO  TaskSchedulerImpl:60 - Adding task set 5.0 with 1 tasks resource profile 0
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) 
2025-01-02 13:14:39 INFO  Executor:60 - Running task 0.0 in stage 5.0 (TID 5)
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
2025-01-02 13:14:39 INFO  ShuffleBlockFetcherIterator:60 - Started 0 remote fetches in 4 ms
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
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)
2025-01-02 13:14:39 INFO  TaskSchedulerImpl:60 - Removed TaskSet 5.0, whose tasks have all completed, from pool 
2025-01-02 13:14:39 INFO  DAGScheduler:60 - ResultStage 5 (showString at NativeMethodAccessorImpl.java:0) finished in 0.077 s
2025-01-02 13:14:39 INFO  DAGScheduler:60 - Job 3 is finished. Cancelling potential speculative or zombie tasks for this job
2025-01-02 13:14:39 INFO  TaskSchedulerImpl:60 - Killing all running tasks in stage 5: Stage finished
2025-01-02 13:14:39 INFO  DAGScheduler:60 - Job 3 finished: showString at NativeMethodAccessorImpl.java:0, took 0.100031 s
+----------+----------+
|year(date)|avg(value)|
+----------+----------+
|      2025|       2.1|
|      2020|       3.0|
+----------+----------+

2025-01-02 13:14:39 INFO  SparkContext:60 - SparkContext is stopping with exitCode 0.
2025-01-02 13:14:39 INFO  AbstractConnector:383 - Stopped Spark@70fe4335{HTTP/1.1, (http/1.1)}{0.0.0.0:4041}
2025-01-02 13:14:39 INFO  SparkUI:60 - Stopped Spark web UI at http://82b3b0c373a8:4041
2025-01-02 13:14:39 INFO  MapOutputTrackerMasterEndpoint:60 - MapOutputTrackerMasterEndpoint stopped!
2025-01-02 13:14:39 INFO  MemoryStore:60 - MemoryStore cleared
2025-01-02 13:14:39 INFO  BlockManager:60 - BlockManager stopped
2025-01-02 13:14:39 INFO  BlockManagerMaster:60 - BlockManagerMaster stopped
2025-01-02 13:14:39 INFO  OutputCommitCoordinator$OutputCommitCoordinatorEndpoint:60 - OutputCommitCoordinator stopped!
2025-01-02 13:14:39 INFO  SparkContext:60 - Successfully stopped SparkContext
INFO:py4j.clientserver:Closing down clientserver connection
2025-01-02 13:14:40 INFO  ShutdownHookManager:60 - Shutdown hook called
2025-01-02 13:14:40 INFO  ShutdownHookManager:60 - Deleting directory /tmp/spark-dafff29b-fd92-4d4e-a958-7575b0ae52a6
2025-01-02 13:14:40 INFO  ShutdownHookManager:60 - Deleting directory /tmp/spark-dafff29b-fd92-4d4e-a958-7575b0ae52a6/pyspark-7626ba41-9b34-4a2b-abcf-67892e474185
2025-01-02 13:14:40 INFO  ShutdownHookManager:60 - Deleting directory /tmp/spark-ea02e098-5cac-4055-b258-09f66acc97d1

Add your own logging messages¶

PySpark's logging system is based on the Log4j logger and is configured in the log4j2.properties file.

You can set up your own logging system and integrate it with PySpark's logging.

We are going to showcase two scenarios:

two distinct logging systems (PySpark's logging system and Python's logging)
unified logging by redirecting PySpark's logs to Python's logging module

Two logging systems¶

In [28]:

%%writefile test.py
from pyspark.sql import SparkSession
import logging

# Create a SparkSession
spark = SparkSession.builder \
   .appName("Logging Demo: two systems") \
   .getOrCreate()

# create my logger and set log level to WARN
logging.basicConfig(level=logging.WARN)
logger = logging.getLogger(__name__)

# set PySpark log level to WARN
sc = spark.sparkContext
sc.setLogLevel("WARN")

rdd = spark.sparkContext.parallelize(range(10**5))
logger.error("Computed sum: %s", rdd.sum())

# Stop the SparkSession
spark.stop()

Writing test.py

In [29]:

!spark-submit test.py

2025-01-02 13:14:45 WARN  NativeCodeLoader:60 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2025-01-02 13:14:47 WARN  Utils:72 - Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
ERROR:__main__:Computed sum: 4999950000

Unified logging¶

In [30]:

%%writefile test.py
from pyspark.sql import SparkSession
import logging

# Create a SparkSession
spark = SparkSession.builder \
   .appName("Logging Demo: unified log system") \
   .getOrCreate()

sc = spark.sparkContext

# use PySpark's logger and set log level to WARN
log4jLogger = sc._jvm.org.apache.log4j
logger = log4jLogger.LogManager.getLogger(__name__)
logger.setLevel(log4jLogger.Level.WARN)  # Set desired logging level

# set PySpark log level to WARN
sc.setLogLevel("WARN")

rdd = spark.sparkContext.parallelize(range(10**5))
logger.warn(f"Computed sum: {rdd.sum()}")

# Stop the SparkSession
spark.stop()

Overwriting test.py

In [31]:

!spark-submit test.py

2025-01-02 13:14:57 WARN  NativeCodeLoader:60 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2025-01-02 13:14:59 WARN  Utils:72 - Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
2025-01-02 13:15:03 WARN  __main__:244 - Computed sum: 4999950000

Now your warning generated by the line

logger.warn(f"Computed sum: {rdd.sum()}")

is integrated with the Log4j messages and has the same format.

PySpark miscellanea¶

Table of contents¶

Preliminaries¶

How to get your application's ID¶

From the Spark session¶

In the PySpark shell¶

Default parallelism¶

What is spark.default.parallelism?¶

Get or set default parallelism¶

Example of how parallelism can influence the runtime of computation¶

How to change PySpark's log level¶

In the PySpark shell¶

In a PySpark script¶

Add your own logging messages¶

Two logging systems¶

Unified logging¶

What is `spark.default.parallelism`?¶