PySpark Documentation

PySpark is a set of Spark APIs in Python language. It not only offers for you to write an application with Python APIs but also provides PySpark shell so you can interactively analyze your data in a distributed environment. PySpark includes almost all Apache Spark features.

General Execution: Spark Core

Spark Core is the underlying general execution engine for the Spark platform that all other functionality is built on top of. It provides in-memory computing capabilities.

Structured Data: Spark SQL

Spark SQL is a Spark module for structured data processing. It provides a programming abstraction called DataFrames and can also act as distributed SQL query engine.

Streaming Analytics: Spark Streaming

Running on top of Spark, Spark Streaming enables powerful interactive and analytical applications across both streaming and historical data, while inheriting Spark’s ease of use and fault tolerance characteristics.

Machine Learning: MLlib

Machine learning has quickly emerged as a critical piece in mining Big Data for actionable insights. Built on top of Spark, MLlib is a scalable machine learning library that delivers both high-quality algorithms (e.g., multiple iterations to increase accuracy) and blazing speed (up to 100x faster than MapReduce).

Sitemap

Getting Started

Quickstart

This tutorial provides a quick introduction to using Spark. We will first introduce the API through Spark’s interactive shell (in Python or Scala), then show how to write applications in Java, Scala, and Python.

To follow along with this guide, first, download a packaged release of Spark from the Spark website. Since we won’t be using HDFS, you can download a package for any version of Hadoop.

Note that, before Spark 2.0, the main programming interface of Spark was the Resilient Distributed Dataset (RDD). After Spark 2.0, RDDs are replaced by Dataset, which is strongly-typed like an RDD, but with richer optimizations under the hood. The RDD interface is still supported, and you can get a more detailed reference at the RDD programming guide. However, we highly recommend you to switch to use Dataset, which has better performance than RDD. See the SQL programming guide to get more information about Dataset.

Interactive Analysis with the Spark Shell

Basics

Spark’s shell provides a simple way to learn the API, as well as a powerful tool to analyze data interactively. It is available in either Scala (which runs on the Java VM and is thus a good way to use existing Java libraries) or Python. Start it by running the following in the Spark directory:

./bin/pyspark

Or if PySpark is installed with pip in your current environment:

pyspark

Spark’s primary abstraction is a distributed collection of items called a Dataset. Datasets can be created from Hadoop InputFormats (such as HDFS files) or by transforming other Datasets. Due to Python’s dynamic nature, we don’t need the Dataset to be strongly-typed in Python. As a result, all Datasets in Python are Dataset[Row], and we call it DataFrame to be consistent with the data frame concept in Pandas and R. Let’s make a new DataFrame from the text of the README file in the Spark source directory:

>>> textFile = spark.read.text("README.md")

You can get values from DataFrame directly, by calling some actions, or transform the DataFrame to get a new one. For more details, please read the _[API doc](api/python/index.html#pyspark.sql.DataFrame)_.

>>> textFile.count()  # Number of rows in this DataFrame
126

>>> textFile.first()  # First row in this DataFrame
Row(value=u'# Apache Spark')

Now let’s transform this DataFrame to a new one. We call filter to return a new DataFrame with a subset of the lines in the file.

>>> linesWithSpark = textFile.filter(textFile.value.contains("Spark"))

We can chain together transformations and actions:

>>> textFile.filter(textFile.value.contains("Spark")).count()  # How many lines contain "Spark"?
15

More on Dataset Operations

Dataset actions and transformations can be used for more complex computations. Let’s say we want to find the line with the most words:

>>> from pyspark.sql.functions import *
>>> textFile.select(size(split(textFile.value, "\s+")).name("numWords")).agg(max(col("numWords"))).collect()
[Row(max(numWords)=15)]

This first maps a line to an integer value and aliases it as “numWords”, creating a new DataFrame. agg is called on that DataFrame to find the largest word count. The arguments to select and agg are both Column, we can use df.colName to get a column from a DataFrame. We can also import pyspark.sql.functions, which provides a lot of convenient functions to build a new Column from an old one.

One common data flow pattern is MapReduce, as popularized by Hadoop. Spark can implement MapReduce flows easily:

>>> wordCounts = textFile.select(explode(split(textFile.value, "\s+")).alias("word")).groupBy("word").count()

Here, we use the explode function in select, to transform a Dataset of lines to a Dataset of words, and then combine groupBy and count to compute the per-word counts in the file as a DataFrame of 2 columns: “word” and “count”. To collect the word counts in our shell, we can call collect:

>>> wordCounts.collect()
[Row(word=u'online', count=1), Row(word=u'graphs', count=1), ...]

Caching

Spark also supports pulling data sets into a cluster-wide in-memory cache. This is very useful when data is accessed repeatedly, such as when querying a small “hot” dataset or when running an iterative algorithm like PageRank. As a simple example, let’s mark our linesWithSpark dataset to be cached:

>>> linesWithSpark.cache()
>>> linesWithSpark.count()
15

>>> linesWithSpark.count()
15

It may seem silly to use Spark to explore and cache a 100-line text file. The interesting part is that these same functions can be used on very large data sets, even when they are striped across tens or hundreds of nodes. You can also do this interactively by connecting bin/pyspark to a cluster, as described in the RDD programming guide.

Self-contained Application

Now we will show how to write an application using the Python API (PySpark).

If you are building a packaged PySpark application or library you can add it to your setup.py file as:

install_requires=[
    'pyspark=={site.SPARK_VERSION}'
]

As an example, we’ll create a simple Spark application, SimpleApp.py:

"""SimpleApp.py"""
from pyspark.sql import SparkSession

logFile = "YOUR_SPARK_HOME/README.md"  # Should be some file on your system
spark = SparkSession.builder.appName("SimpleApp").getOrCreate()
logData = spark.read.text(logFile).cache()

numAs = logData.filter(logData.value.contains('a')).count()
numBs = logData.filter(logData.value.contains('b')).count()

print("Lines with a: %i, lines with b: %i" % (numAs, numBs))

spark.stop()

This program just counts the number of lines containing ‘a’ and the number containing ‘b’ in a text file. Note that you’ll need to replace YOUR_SPARK_HOME with the location where Spark is installed. As with the Scala and Java examples, we use a SparkSession to create Datasets. For applications that use custom classes or third-party libraries, we can also add code dependencies to spark-submit through its –py-files argument by packaging them into a .zip file (see spark-submit –help for details). SimpleApp is simple enough that we do not need to specify any code dependencies.

We can run this application using the bin/spark-submit script:

# Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit \
  --master local[4] \
  SimpleApp.py
...
Lines with a: 46, Lines with b: 23

If you have PySpark pip installed into your environment (e.g., pip install pyspark), you can run your application with the regular Python interpreter or use the provided ‘spark-submit’ as you prefer.

# Use the Python interpreter to run your application
$ python SimpleApp.py
...

Installation

Binary Distribution

PyPi

Enviornments

NumPy, PyArrow and Pandas

Package Overview

Structure

Functions

Basic Functionality

SparkContext

SparkSession

DataFrame

RDD

Reading CSV and Writing JSON

Tutorials

Handling Missing Data

Plotting and Visualization

User Guide

PySpark Usage Guide for Pandas with Apache Arrow

Apache Arrow in PySpark

Apache Arrow is an in-memory columnar data format that is used in Spark to efficiently transfer data between JVM and Python processes. This currently is most beneficial to Python users that work with Pandas/NumPy data. Its usage is not automatic and might require some minor changes to configuration or code to take full advantage and ensure compatibility. This guide will give a high-level description of how to use Arrow in Spark and highlight any differences when working with Arrow-enabled data.

Ensure PyArrow Installed

To use Apache Arrow in PySpark, the recommended version of PyArrow should be installed. If you install PySpark using pip, then PyArrow can be brought in as an extra dependency of the SQL module with the command pip install pyspark[sql]. Otherwise, you must ensure that PyArrow is installed and available on all cluster nodes. You can install using pip or conda from the conda-forge channel. See PyArrow installation for details.

Enabling for Conversion to/from Pandas

Arrow is available as an optimization when converting a Spark DataFrame to a Pandas DataFrame using the call toPandas() and when creating a Spark DataFrame from a Pandas DataFrame with createDataFrame(pandas_df). To use Arrow when executing these calls, users need to first set the Spark configuration spark.sql.execution.arrow.pyspark.enabled to true. This is disabled by default.

In addition, optimizations enabled by spark.sql.execution.arrow.pyspark.enabled could fallback automatically to non-Arrow optimization implementation if an error occurs before the actual computation within Spark. This can be controlled by spark.sql.execution.arrow.pyspark.fallback.enabled.

import numpy as np
import pandas as pd

# Enable Arrow-based columnar data transfers
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")

# Generate a Pandas DataFrame
pdf = pd.DataFrame(np.random.rand(100, 3))

# Create a Spark DataFrame from a Pandas DataFrame using Arrow
df = spark.createDataFrame(pdf)

# Convert the Spark DataFrame back to a Pandas DataFrame using Arrow
result_pdf = df.select("*").toPandas()

Using the above optimizations with Arrow will produce the same results as when Arrow is not enabled. Note that even with Arrow, toPandas() results in the collection of all records in the DataFrame to the driver program and should be done on a small subset of the data. Not all Spark data types are currently supported and an error can be raised if a column has an unsupported type, see Supported SQL Types. If an error occurs during createDataFrame(), Spark will fall back to create the DataFrame without Arrow.

Pandas UDFs (a.k.a. Vectorized UDFs)

Pandas UDFs are user defined functions that are executed by Spark using Arrow to transfer data and Pandas to work with the data, which allows vectorized operations. A Pandas UDF is defined using the pandas_udf as a decorator or to wrap the function, and no additional configuration is required. A Pandas UDF behaves as a regular PySpark function API in general.

Before Spark 3.0, Pandas UDFs used to be defined with PandasUDFType. From Spark 3.0 with Python 3.6+, you can also use Python type hints. Using Python type hints are preferred and using PandasUDFType will be deprecated in the future release.

Note that the type hint should use pandas.Series in all cases but there is one variant that pandas.DataFrame should be used for its input or output type hint instead when the input or output column is of StructType. The following example shows a Pandas UDF which takes long column, string column and struct column, and outputs a struct column. It requires the function to specify the type hints of pandas.Series and pandas.DataFrame as below:

import pandas as pd

from pyspark.sql.functions import pandas_udf

@pandas_udf("col1 string, col2 long")
def func(s1: pd.Series, s2: pd.Series, s3: pd.DataFrame) -> pd.DataFrame:
    s3['col2'] = s1 + s2.str.len()
    return s3

# Create a Spark DataFrame that has three columns including a sturct column.
df = spark.createDataFrame(
    [[1, "a string", ("a nested string",)]],
    "long_col long, string_col string, struct_col struct<col1:string>")

df.printSchema()
# root
# |-- long_column: long (nullable = true)
# |-- string_column: string (nullable = true)
# |-- struct_column: struct (nullable = true)
# |    |-- col1: string (nullable = true)

df.select(func("long_col", "string_col", "struct_col")).printSchema()
# |-- func(long_col, string_col, struct_col): struct (nullable = true)
# |    |-- col1: string (nullable = true)
# |    |-- col2: long (nullable = true)

In the following sections, it describes the combinations of the supported type hints. For simplicity, pandas.DataFrame variant is omitted.

Series to Series

The type hint can be expressed as pandas.Series, … -> pandas.Series.

By using pandas_udf with the function having such type hints above, it creates a Pandas UDF where the given function takes one or more pandas.Series and outputs one pandas.Series. The output of the function should always be of the same length as the input. Internally, PySpark will execute a Pandas UDF by splitting columns into batches and calling the function for each batch as a subset of the data, then concatenating the results together.

The following example shows how to create this Pandas UDF that computes the product of 2 columns.

import pandas as pd

from pyspark.sql.functions import col, pandas_udf
from pyspark.sql.types import LongType

# Declare the function and create the UDF
def multiply_func(a: pd.Series, b: pd.Series) -> pd.Series:
    return a * b

multiply = pandas_udf(multiply_func, returnType=LongType())

# The function for a pandas_udf should be able to execute with local Pandas data
x = pd.Series([1, 2, 3])
print(multiply_func(x, x))
# 0    1
# 1    4
# 2    9
# dtype: int64

# Create a Spark DataFrame, 'spark' is an existing SparkSession
df = spark.createDataFrame(pd.DataFrame(x, columns=["x"]))

# Execute function as a Spark vectorized UDF
df.select(multiply(col("x"), col("x"))).show()
# +-------------------+
# |multiply_func(x, x)|
# +-------------------+
# |                  1|
# |                  4|
# |                  9|
# +-------------------+

For detailed usage, please see `pyspark.sql.functions.pandas_udf.

Iterator of Series to Iterator of Series

The type hint can be expressed as Iterator[pandas.Series] -> Iterator[pandas.Series].

By using pandas_udf with the function having such type hints above, it creates a Pandas UDF where the given function takes an iterator of pandas.Series and outputs an iterator of pandas.Series. The length of the entire output from the function should be the same length of the entire input; therefore, it can prefetch the data from the input iterator as long as the lengths are the same. In this case, the created Pandas UDF requires one input column when the Pandas UDF is called. To use multiple input columns, a different type hint is required. See Iterator of Multiple Series to Iterator of Series.

It is also useful when the UDF execution requires initializing some states although internally it works identically as Series to Series case. The pseudocode below illustrates the example.

{% highlight python %} @pandas_udf(“long”) def calculate(iterator: Iterator[pd.Series]) -> Iterator[pd.Series]: # Do some expensive initialization with a state state = very_expensive_initialization() for x in iterator: # Use that state for whole iterator. yield calculate_with_state(x, state)

df.select(calculate(“value”)).show() {% endhighlight %}

The following example shows how to create this Pandas UDF:

from typing import Iterator

import pandas as pd

from pyspark.sql.functions import pandas_udf

pdf = pd.DataFrame([1, 2, 3], columns=["x"])
df = spark.createDataFrame(pdf)

# Declare the function and create the UDF
@pandas_udf("long")
def plus_one(iterator: Iterator[pd.Series]) -> Iterator[pd.Series]:
    for x in iterator:
        yield x + 1

df.select(plus_one("x")).show()
# +-----------+
# |plus_one(x)|
# +-----------+
# |          2|
# |          3|
# |          4|
# +-----------+

For detailed usage, please see `pyspark.sql.functions.pandas_udf.

Iterator of Multiple Series to Iterator of Series

The type hint can be expressed as Iterator[Tuple[pandas.Series, ...]] -> Iterator[pandas.Series].

By using pandas_udf with the function having such type hints above, it creates a Pandas UDF where the given function takes an iterator of a tuple of multiple pandas.Series and outputs an iterator of pandas.Series. In this case, the created pandas UDF requires multiple input columns as many as the series in the tuple when the Pandas UDF is called. Otherwise, it has the same characteristics and restrictions as Iterator of Series to Iterator of Series case.

The following example shows how to create this Pandas UDF:

from typing import Iterator, Tuple

import pandas as pd

from pyspark.sql.functions import pandas_udf

pdf = pd.DataFrame([1, 2, 3], columns=["x"])
df = spark.createDataFrame(pdf)

# Declare the function and create the UDF
@pandas_udf("long")
def multiply_two_cols(
        iterator: Iterator[Tuple[pd.Series, pd.Series]]) -> Iterator[pd.Series]:
    for a, b in iterator:
        yield a * b

df.select(multiply_two_cols("x", "x")).show()
# +-----------------------+
# |multiply_two_cols(x, x)|
# +-----------------------+
# |                      1|
# |                      4|
# |                      9|
# +-----------------------+

For detailed usage, please see `pyspark.sql.functions.pandas_udf.

Series to Scalar

The type hint can be expressed as pandas.Series, … -> Any.

By using pandas_udf with the function having such type hints above, it creates a Pandas UDF similar to PySpark’s aggregate functions. The given function takes pandas.Series and returns a scalar value. The return type should be a primitive data type, and the returned scalar can be either a python primitive type, e.g., int or float or a numpy data type, e.g., numpy.int64 or numpy.float64. Any should ideally be a specific scalar type accordingly.

This UDF can be also used with groupBy().agg() and `pyspark.sql.Window <api/python/pyspark.sql.html#pyspark.sql.Window>`__. It defines an aggregation from one or more pandas.Series to a scalar value, where each pandas.Series represents a column within the group or window.

Note that this type of UDF does not support partial aggregation and all data for a group or window will be loaded into memory. Also, only unbounded window is supported with Grouped aggregate Pandas UDFs currently. The following example shows how to use this type of UDF to compute mean with a group-by and window operations:

import pandas as pd

from pyspark.sql.functions import pandas_udf
from pyspark.sql import Window

df = spark.createDataFrame(
    [(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)],
    ("id", "v"))

# Declare the function and create the UDF
@pandas_udf("double")
def mean_udf(v: pd.Series) -> float:
    return v.mean()

df.select(mean_udf(df['v'])).show()
# +-----------+
# |mean_udf(v)|
# +-----------+
# |        4.2|
# +-----------+

df.groupby("id").agg(mean_udf(df['v'])).show()
# +---+-----------+
# | id|mean_udf(v)|
# +---+-----------+
# |  1|        1.5|
# |  2|        6.0|
# +---+-----------+

w = Window \
    .partitionBy('id') \
    .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
df.withColumn('mean_v', mean_udf(df['v']).over(w)).show()
# +---+----+------+
# | id|   v|mean_v|
# +---+----+------+
# |  1| 1.0|   1.5|
# |  1| 2.0|   1.5|
# |  2| 3.0|   6.0|
# |  2| 5.0|   6.0|
# |  2|10.0|   6.0|
# +---+----+------+

For detailed usage, please see `pyspark.sql.functions.pandas_udf.

Pandas Function APIs

Pandas Function APIs can directly apply a Python native function against the whole DataFrame by using Pandas instances. Internally it works similarly with Pandas UDFs by using Arrow to transfer data and Pandas to work with the data, which allows vectorized operations. However, A Pandas Function API behaves as a regular API under PySpark DataFrame instead of Column, and Python type hints in Pandas Functions APIs are optional and do not affect how it works internally at this moment although they might be required in the future.

From Spark 3.0, grouped map pandas UDF is now categorized as a separate Pandas Function API, DataFrame.groupby().applyInPandas(). It is still possible to use it with PandasUDFType and DataFrame.groupby().apply() as it was; however, it is preferred to use DataFrame.groupby().applyInPandas() directly. Using PandasUDFType will be deprecated in the future.

Grouped Map

Grouped map operations with Pandas instances are supported by DataFrame.groupby().applyInPandas() which requires a Python function that takes a pandas.DataFrame and return another pandas.DataFrame. It maps each group to each pandas.DataFrame in the Python function.

This API implements the “split-apply-combine” pattern which consists of three steps:

• Split the data into groups by using DataFrame.groupBy.

• Apply a function on each group. The input and output of the function are both pandas.DataFrame. The input data contains all the rows and columns for each group.

• Combine the results into a new PySpark DataFrame.

To use groupBy().applyInPandas(), the user needs to define the following:

• A Python function that defines the computation for each group.

• A StructType object or a string that defines the schema of the output PySpark DataFrame.

The column labels of the returned pandas.DataFrame must either match the field names in the defined output schema if specified as strings, or match the field data types by position if not strings, e.g. integer indices. See pandas.DataFrame on how to label columns when constructing a pandas.DataFrame.

Note that all data for a group will be loaded into memory before the function is applied. This can lead to out of memory exceptions, especially if the group sizes are skewed. The configuration for maxRecordsPerBatch is not applied on groups and it is up to the user to ensure that the grouped data will fit into the available memory.

The following example shows how to use groupby().applyInPandas() to subtract the mean from each value in the group.

df = spark.createDataFrame(
    [(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)],
    ("id", "v"))

def subtract_mean(pdf):
    # pdf is a pandas.DataFrame
    v = pdf.v
    return pdf.assign(v=v - v.mean())

df.groupby("id").applyInPandas(subtract_mean, schema="id long, v double").show()
# +---+----+
# | id|   v|
# +---+----+
# |  1|-0.5|
# |  1| 0.5|
# |  2|-3.0|
# |  2|-1.0|
# |  2| 4.0|
# +---+----+

For detailed usage, please see `pyspark.sql.GroupedData.applyInPandas.

Map

Map operations with Pandas instances are supported by DataFrame.mapInPandas() which maps an iterator of pandas.DataFrames to another iterator of pandas.DataFrame that represents the current PySpark DataFrame and returns the result as a PySpark DataFrame. The functions takes and outputs an iterator of pandas.DataFrame. It can return the output of arbitrary length in contrast to some Pandas UDFs although internally it works similarly with Series to Series Pandas UDF.

The following example shows how to use mapInPandas():

df = spark.createDataFrame([(1, 21), (2, 30)], ("id", "age"))

def filter_func(iterator):
    for pdf in iterator:
        yield pdf[pdf.id == 1]

df.mapInPandas(filter_func, schema=df.schema).show()
# +---+---+
# | id|age|
# +---+---+
# |  1| 21|
# +---+---+

For detailed usage, please see `pyspark.sql.DataFrame.mapsInPandas.

Co-grouped Map

Co-grouped map operations with Pandas instances are supported by DataFrame.groupby().cogroup().applyInPandas() which allows two PySpark DataFrames to be cogrouped by a common key and then a Python function applied to each cogroup. It consists of the following steps: * Shuffle the data such that the groups of each dataframe which share a key are cogrouped together. * Apply a function to each cogroup. The input of the function is two pandas.DataFrame (with an optional tuple representing the key). The output of the function is a pandas.DataFrame. * Combine the pandas.DataFrames from all groups into a new PySpark DataFrame.

To use groupBy().cogroup().applyInPandas(), the user needs to define the following: * A Python function that defines the computation for each cogroup. * A StructType object or a string that defines the schema of the output PySpark DataFrame.

The column labels of the returned pandas.DataFrame must either match the field names in the defined output schema if specified as strings, or match the field data types by position if not strings, e.g. integer indices. See pandas.DataFrame on how to label columns when constructing a pandas.DataFrame.

Note that all data for a cogroup will be loaded into memory before the function is applied. This can lead to out of memory exceptions, especially if the group sizes are skewed. The configuration for maxRecordsPerBatch is not applied and it is up to the user to ensure that the cogrouped data will fit into the available memory.

The following example shows how to use groupby().cogroup().applyInPandas() to perform an asof join between two datasets.

import pandas as pd

df1 = spark.createDataFrame(
    [(20000101, 1, 1.0), (20000101, 2, 2.0), (20000102, 1, 3.0), (20000102, 2, 4.0)],
    ("time", "id", "v1"))

df2 = spark.createDataFrame(
    [(20000101, 1, "x"), (20000101, 2, "y")],
    ("time", "id", "v2"))

def asof_join(l, r):
    return pd.merge_asof(l, r, on="time", by="id")

df1.groupby("id").cogroup(df2.groupby("id")).applyInPandas(
    asof_join, schema="time int, id int, v1 double, v2 string").show()
# +--------+---+---+---+
# |    time| id| v1| v2|
# +--------+---+---+---+
# |20000101|  1|1.0|  x|
# |20000102|  1|3.0|  x|
# |20000101|  2|2.0|  y|
# |20000102|  2|4.0|  y|
# +--------+---+---+---+

For detailed usage, please see `pyspark.sql.PandasCogroupedOps.applyInPandas().

Usage Notes

Supported SQL Types

Currently, all Spark SQL data types are supported by Arrow-based conversion except MapType, ArrayType of TimestampType, and nested StructType.

Setting Arrow Batch Size

Data partitions in Spark are converted into Arrow record batches, which can temporarily lead to high memory usage in the JVM. To avoid possible out of memory exceptions, the size of the Arrow record batches can be adjusted by setting the conf “spark.sql.execution.arrow.maxRecordsPerBatch” to an integer that will determine the maximum number of rows for each batch. The default value is 10,000 records per batch. If the number of columns is large, the value should be adjusted accordingly. Using this limit, each data partition will be made into 1 or more record batches for processing.

Timestamp with Time Zone Semantics

Spark internally stores timestamps as UTC values, and timestamp data that is brought in without a specified time zone is converted as local time to UTC with microsecond resolution. When timestamp data is exported or displayed in Spark, the session time zone is used to localize the timestamp values. The session time zone is set with the configuration ‘spark.sql.session.timeZone’ and will default to the JVM system local time zone if not set. Pandas uses a datetime64 type with nanosecond resolution, datetime64[ns], with optional time zone on a per-column basis.

When timestamp data is transferred from Spark to Pandas it will be converted to nanoseconds and each column will be converted to the Spark session time zone then localized to that time zone, which removes the time zone and displays values as local time. This will occur when calling toPandas() or pandas_udf with timestamp columns.

When timestamp data is transferred from Pandas to Spark, it will be converted to UTC microseconds. This occurs when calling createDataFrame with a Pandas DataFrame or when returning a timestamp from a pandas_udf. These conversions are done automatically to ensure Spark will have data in the expected format, so it is not necessary to do any of these conversions yourself. Any nanosecond values will be truncated.

Note that a standard UDF (non-Pandas) will load timestamp data as Python datetime objects, which is different than a Pandas timestamp. It is recommended to use Pandas time series functionality when working with timestamps in pandas_udfs to get the best performance, see here for details.

Recommended Pandas and PyArrow Versions

For usage with pyspark.sql, the supported versions of Pandas is 0.24.2 and PyArrow is 0.15.1. Higher versions may be used, however, compatibility and data correctness can not be guaranteed and should be verified by the user.

Compatibility Setting for PyArrow >= 0.15.0 and Spark 2.3.x, 2.4.x

Since Arrow 0.15.0, a change in the binary IPC format requires an environment variable to be compatible with previous versions of Arrow <= 0.14.1. This is only necessary to do for PySpark users with versions 2.3.x and 2.4.x that have manually upgraded PyArrow to 0.15.0. The following can be added to conf/spark-env.sh to use the legacy Arrow IPC format:

ARROW_PRE_0_15_IPC_FORMAT=1

This will instruct PyArrow >= 0.15.0 to use the legacy IPC format with the older Arrow Java that is in Spark 2.3.x and 2.4.x. Not setting this environment variable will lead to a similar error as described in SPARK-29367 when running pandas_udfs or toPandas() with Arrow enabled. More information about the Arrow IPC change can be read on the Arrow 0.15.0 release blog.

Pandas UDFs with Type Hints

Python Type Hints

Pandas UDFs In Spark 2.4

Pandas UDFs

Pandas Function APIs

Working with CSV and JSON

Reading CSV

Transformation in SQL

Writing as JSON —————a

Configurations

Core Configurations

SQL Configurations

PyArrow Configurations

Monitoring

Python Profiler

Python Debugger

Monitoring Python Workers

API Reference

This page lists an overview of all public PySpark modules, classes, functions and methods.

Spark SQL

Spark SQL is a Spark module for structured data processing. Unlike the basic Spark RDD API, the interfaces provided by Spark SQL provide Spark with more information about the structure of both the data and the computation being performed. Internally, Spark SQL uses this extra information to perform extra optimizations. There are several ways to interact with Spark SQL including SQL and the Dataset API. When computing a result the same execution engine is used, independent of which API/language you are using to express the computation.

This unification means that developers can easily switch back and forth between different APIs based on which provides the most natural way to express a given transformation.

Core Classes

The classes below are core classes in Spark SQL APIs at PySpark.

SparkSession(sparkContext[, jsparkSession])	The entry point to programming Spark with the Dataset and DataFrame API.
DataFrame(jdf, sql_ctx)	A distributed collection of data grouped into named columns.
Column(jc)	A column in a DataFrame.
Row	A row in DataFrame.
GroupedData(jgd, df)	A set of methods for aggregations on a DataFrame, created by DataFrame.groupBy().
DataFrameNaFunctions(df)	Functionality for working with missing data in DataFrame.
DataFrameStatFunctions(df)	Functionality for statistic functions with DataFrame.
Window	Utility functions for defining window in DataFrames.

Spark Session APIs

The entry point to programming Spark with the Dataset and DataFrame API.

SparkSession.builder	A class attribute having a Builder to construct SparkSession instances.
SparkSession.builder.appName(name)	Sets a name for the application, which will be shown in the Spark web UI.
SparkSession.builder.config([key, value, conf])	Sets a config option.
SparkSession.builder.enableHiveSupport()	Enables Hive support, including connectivity to a persistent Hive metastore, support for Hive SerDes, and Hive user-defined functions.
SparkSession.builder.getOrCreate()	Gets an existing SparkSession or, if there is no existing one, creates a new one based on the options set in this builder.
SparkSession.builder.master(master)	Sets the Spark master URL to connect to, such as “local” to run locally, “local[4]” to run locally with 4 cores, or “spark://master:7077” to run on a Spark standalone cluster.
SparkSession.catalog	Interface through which the user may create, drop, alter or query underlying databases, tables, functions, etc.
SparkSession.conf	Runtime configuration interface for Spark.
SparkSession.createDataFrame(data[, schema, …])	Creates a DataFrame from an RDD, a list or a pandas.DataFrame.
SparkSession.getActiveSession()	Returns the active SparkSession for the current thread, returned by the builder.
SparkSession.newSession()	Returns a new SparkSession as new session, that has separate SQLConf, registered temporary views and UDFs, but shared SparkContext and table cache.
SparkSession.range(start[, end, step, …])	Create a DataFrame with single pyspark.sql.types.LongType column named id, containing elements in a range from start to end (exclusive) with step value step.
SparkSession.read	Returns a DataFrameReader that can be used to read data in as a DataFrame.
SparkSession.readStream	Returns a DataStreamReader that can be used to read data streams as a streaming DataFrame.
SparkSession.sparkContext	Returns the underlying SparkContext.
SparkSession.sql(sqlQuery)	Returns a DataFrame representing the result of the given query.
SparkSession.stop()	Stop the underlying SparkContext.
SparkSession.streams	Returns a StreamingQueryManager that allows managing all the StreamingQuery instances active on this context.
SparkSession.table(tableName)	Returns the specified table as a DataFrame.
SparkSession.udf	Returns a UDFRegistration for UDF registration.
SparkSession.version	The version of Spark on which this application is running.

Input and Output

DataFrameReader.csv(path[, schema, sep, …])	Loads a CSV file and returns the result as a DataFrame.
DataFrameReader.format(source)	Specifies the input data source format.
DataFrameReader.jdbc(url, table[, column, …])	Construct a DataFrame representing the database table named table accessible via JDBC URL url and connection properties.
DataFrameReader.json(path[, schema, …])	Loads JSON files and returns the results as a DataFrame.
DataFrameReader.load([path, format, schema])	Loads data from a data source and returns it as a DataFrame.
DataFrameReader.option(key, value)	Adds an input option for the underlying data source.
DataFrameReader.options(**options)	Adds input options for the underlying data source.
DataFrameReader.orc(path[, mergeSchema, …])	Loads ORC files, returning the result as a DataFrame.
DataFrameReader.parquet(*paths, **options)	Loads Parquet files, returning the result as a DataFrame.
DataFrameReader.schema(schema)	Specifies the input schema.
DataFrameReader.table(tableName)	Returns the specified table as a DataFrame.
DataFrameReader.text(paths[, wholetext, …])	Loads text files and returns a DataFrame whose schema starts with a string column named “value”, and followed by partitioned columns if there are any.
DataFrameWriter.bucketBy(numBuckets, col, *cols)	Buckets the output by the given columns.If specified, the output is laid out on the file system similar to Hive’s bucketing scheme.
DataFrameWriter.csv(path[, mode, …])	Saves the content of the DataFrame in CSV format at the specified path.
DataFrameWriter.format(source)	Specifies the underlying output data source.
DataFrameWriter.insertInto(tableName[, …])	Inserts the content of the DataFrame to the specified table.
DataFrameWriter.jdbc(url, table[, mode, …])	Saves the content of the DataFrame to an external database table via JDBC.
DataFrameWriter.json(path[, mode, …])	Saves the content of the DataFrame in JSON format (JSON Lines text format or newline-delimited JSON) at the specified path.
DataFrameWriter.mode(saveMode)	Specifies the behavior when data or table already exists.
DataFrameWriter.option(key, value)	Adds an output option for the underlying data source.
DataFrameWriter.options(**options)	Adds output options for the underlying data source.
DataFrameWriter.orc(path[, mode, …])	Saves the content of the DataFrame in ORC format at the specified path.
DataFrameWriter.parquet(path[, mode, …])	Saves the content of the DataFrame in Parquet format at the specified path.
DataFrameWriter.partitionBy(*cols)	Partitions the output by the given columns on the file system.
DataFrameWriter.save([path, format, mode, …])	Saves the contents of the DataFrame to a data source.
DataFrameWriter.saveAsTable(name[, format, …])	Saves the content of the DataFrame as the specified table.
DataFrameWriter.sortBy(col, *cols)	Sorts the output in each bucket by the given columns on the file system.
DataFrameWriter.text(path[, compression, …])	Saves the content of the DataFrame in a text file at the specified path.

DataFrame APIs

DataFrame.agg(*exprs)	Aggregate on the entire DataFrame without groups (shorthand for df.groupBy.agg()).
DataFrame.alias(alias)	Returns a new DataFrame with an alias set.
DataFrame.approxQuantile(col, probabilities, …)	Calculates the approximate quantiles of numerical columns of a DataFrame.
DataFrame.cache()	Persists the DataFrame with the default storage level (MEMORY_AND_DISK).
DataFrame.checkpoint([eager])	Returns a checkpointed version of this Dataset.
DataFrame.coalesce(numPartitions)	Returns a new DataFrame that has exactly numPartitions partitions.
DataFrame.colRegex(colName)	Selects column based on the column name specified as a regex and returns it as Column.
DataFrame.collect()	Returns all the records as a list of Row.
DataFrame.columns	Returns all column names as a list.
DataFrame.corr(col1, col2[, method])	Calculates the correlation of two columns of a DataFrame as a double value.
DataFrame.count()	Returns the number of rows in this DataFrame.
DataFrame.cov(col1, col2)	Calculate the sample covariance for the given columns, specified by their names, as a double value.
DataFrame.createGlobalTempView(name)	Creates a global temporary view with this DataFrame.
DataFrame.createOrReplaceGlobalTempView(name)	Creates or replaces a global temporary view using the given name.
DataFrame.createOrReplaceTempView(name)	Creates or replaces a local temporary view with this DataFrame.
DataFrame.createTempView(name)	Creates a local temporary view with this DataFrame.
DataFrame.crossJoin(other)	Returns the cartesian product with another DataFrame.
DataFrame.crosstab(col1, col2)	Computes a pair-wise frequency table of the given columns.
DataFrame.cube(*cols)	Create a multi-dimensional cube for the current DataFrame using the specified columns, so we can run aggregations on them.
DataFrame.describe(*cols)	Computes basic statistics for numeric and string columns.
DataFrame.distinct()	Returns a new DataFrame containing the distinct rows in this DataFrame.
DataFrame.drop(*cols)	Returns a new DataFrame that drops the specified column.
DataFrame.dropDuplicates([subset])	Return a new DataFrame with duplicate rows removed, optionally only considering certain columns.
DataFrame.drop_duplicates([subset])	drop_duplicates() is an alias for dropDuplicates().
DataFrame.dropna([how, thresh, subset])	Returns a new DataFrame omitting rows with null values.
DataFrame.dtypes	Returns all column names and their data types as a list.
DataFrame.exceptAll(other)	Return a new DataFrame containing rows in this DataFrame but not in another DataFrame while preserving duplicates.
DataFrame.explain([extended, mode])	Prints the (logical and physical) plans to the console for debugging purpose.
DataFrame.fillna(value[, subset])	Replace null values, alias for na.fill().
DataFrame.filter(condition)	Filters rows using the given condition.
DataFrame.first()	Returns the first row as a Row.
DataFrame.foreach(f)	Applies the f function to all Row of this DataFrame.
DataFrame.foreachPartition(f)	Applies the f function to each partition of this DataFrame.
DataFrame.freqItems(cols[, support])	Finding frequent items for columns, possibly with false positives.
DataFrame.groupBy(*cols)	Groups the DataFrame using the specified columns, so we can run aggregation on them.
DataFrame.groupby(*cols)	groupby() is an alias for groupBy().
DataFrame.head([n])	Returns the first n rows.
DataFrame.hint(name, *parameters)	Specifies some hint on the current DataFrame.
DataFrame.inputFiles()	Returns a best-effort snapshot of the files that compose this DataFrame.
DataFrame.intersect(other)	Return a new DataFrame containing rows only in both this DataFrame and another DataFrame.
DataFrame.intersectAll(other)	Return a new DataFrame containing rows in both this DataFrame and another DataFrame while preserving duplicates.
DataFrame.isLocal()	Returns True if the collect() and take() methods can be run locally (without any Spark executors).
DataFrame.isStreaming	Returns True if this Dataset contains one or more sources that continuously return data as it arrives.
DataFrame.join(other[, on, how])	Joins with another DataFrame, using the given join expression.
DataFrame.limit(num)	Limits the result count to the number specified.
DataFrame.localCheckpoint([eager])	Returns a locally checkpointed version of this Dataset.
DataFrame.mapInPandas(func, schema)	Maps an iterator of batches in the current DataFrame using a Python native function that takes and outputs a pandas DataFrame, and returns the result as a DataFrame.
DataFrame.na	Returns a DataFrameNaFunctions for handling missing values.
DataFrame.orderBy(*cols, **kwargs)	Returns a new DataFrame sorted by the specified column(s).
DataFrame.persist([storageLevel])	Sets the storage level to persist the contents of the DataFrame across operations after the first time it is computed.
DataFrame.printSchema()	Prints out the schema in the tree format.
DataFrame.randomSplit(weights[, seed])	Randomly splits this DataFrame with the provided weights.
DataFrame.rdd	Returns the content as an pyspark.RDD of Row.
DataFrame.registerTempTable(name)	Registers this DataFrame as a temporary table using the given name.
DataFrame.repartition(numPartitions, *cols)	Returns a new DataFrame partitioned by the given partitioning expressions.
DataFrame.repartitionByRange(numPartitions, …)	Returns a new DataFrame partitioned by the given partitioning expressions.
DataFrame.replace(to_replace[, value, subset])	Returns a new DataFrame replacing a value with another value.
DataFrame.rollup(*cols)	Create a multi-dimensional rollup for the current DataFrame using the specified columns, so we can run aggregation on them.
DataFrame.sameSemantics(other)	Returns True when the logical query plans inside both DataFrames are equal and therefore return same results.
DataFrame.sample([withReplacement, …])	Returns a sampled subset of this DataFrame.
DataFrame.sampleBy(col, fractions[, seed])	Returns a stratified sample without replacement based on the fraction given on each stratum.
DataFrame.schema	Returns the schema of this DataFrame as a pyspark.sql.types.StructType.
DataFrame.select(*cols)	Projects a set of expressions and returns a new DataFrame.
DataFrame.selectExpr(*expr)	Projects a set of SQL expressions and returns a new DataFrame.
DataFrame.semanticHash()	Returns a hash code of the logical query plan against this DataFrame.
DataFrame.show([n, truncate, vertical])	Prints the first n rows to the console.
DataFrame.sort(*cols, **kwargs)	Returns a new DataFrame sorted by the specified column(s).
DataFrame.sortWithinPartitions(*cols, **kwargs)	Returns a new DataFrame with each partition sorted by the specified column(s).
DataFrame.stat	Returns a DataFrameStatFunctions for statistic functions.
DataFrame.storageLevel	Get the DataFrame’s current storage level.
DataFrame.subtract(other)	Return a new DataFrame containing rows in this DataFrame but not in another DataFrame.
DataFrame.summary(*statistics)	Computes specified statistics for numeric and string columns.
DataFrame.tail(num)	Returns the last num rows as a list of Row.
DataFrame.take(num)	Returns the first num rows as a list of Row.
DataFrame.toDF(*cols)	Returns a new DataFrame that with new specified column names
DataFrame.toJSON([use_unicode])	Converts a DataFrame into a RDD of string.
DataFrame.toLocalIterator([prefetchPartitions])	Returns an iterator that contains all of the rows in this DataFrame.
DataFrame.toPandas()	Returns the contents of this DataFrame as Pandas pandas.DataFrame.
DataFrame.transform(func)	Returns a new DataFrame.
DataFrame.union(other)	Return a new DataFrame containing union of rows in this and another DataFrame.
DataFrame.unionAll(other)	Return a new DataFrame containing union of rows in this and another DataFrame.
DataFrame.unionByName(other)	Returns a new DataFrame containing union of rows in this and another DataFrame.
DataFrame.unpersist([blocking])	Marks the DataFrame as non-persistent, and remove all blocks for it from memory and disk.
DataFrame.where(condition)	where() is an alias for filter().
DataFrame.withColumn(colName, col)	Returns a new DataFrame by adding a column or replacing the existing column that has the same name.
DataFrame.withColumnRenamed(existing, new)	Returns a new DataFrame by renaming an existing column.
DataFrame.withWatermark(eventTime, …)	Defines an event time watermark for this DataFrame.
DataFrame.write	Interface for saving the content of the non-streaming DataFrame out into external storage.
DataFrame.writeStream	Interface for saving the content of the streaming DataFrame out into external storage.
DataFrameNaFunctions.drop([how, thresh, subset])	Returns a new DataFrame omitting rows with null values.
DataFrameNaFunctions.fill(value[, subset])	Replace null values, alias for na.fill().
DataFrameNaFunctions.replace(to_replace[, …])	Returns a new DataFrame replacing a value with another value.
DataFrameStatFunctions.approxQuantile(col, …)	Calculates the approximate quantiles of numerical columns of a DataFrame.
DataFrameStatFunctions.corr(col1, col2[, method])	Calculates the correlation of two columns of a DataFrame as a double value.
DataFrameStatFunctions.cov(col1, col2)	Calculate the sample covariance for the given columns, specified by their names, as a double value.
DataFrameStatFunctions.crosstab(col1, col2)	Computes a pair-wise frequency table of the given columns.
DataFrameStatFunctions.freqItems(cols[, support])	Finding frequent items for columns, possibly with false positives.
DataFrameStatFunctions.sampleBy(col, fractions)	Returns a stratified sample without replacement based on the fraction given on each stratum.

Data Types

ArrayType(elementType[, containsNull])	Array data type.
BinaryType	Binary (byte array) data type.
BooleanType	Boolean data type.
ByteType	Byte data type, i.e.
DataType	Base class for data types.
DateType	Date (datetime.date) data type.
DecimalType([precision, scale])	Decimal (decimal.Decimal) data type.
DoubleType	Double data type, representing double precision floats.
FloatType	Float data type, representing single precision floats.
IntegerType	Int data type, i.e.
LongType	Long data type, i.e.
MapType(keyType, valueType[, valueContainsNull])	Map data type.
NullType	Null type.
Row	A row in DataFrame.
ShortType	Short data type, i.e.
StringType	String data type.
StructField(name, dataType[, nullable, metadata])	A field in StructType.
StructType([fields])	Struct type, consisting of a list of StructField.
TimestampType	Timestamp (datetime.datetime) data type.

Functions

abs(col)	Computes the absolute value.
acos(col)	return inverse cosine of col, as if computed by java.lang.Math.acos()
add_months(start, months)	Returns the date that is months months after start
aggregate(col, zero, merge[, finish])	Applies a binary operator to an initial state and all elements in the array, and reduces this to a single state.
approxCountDistinct(col[, rsd])	Note Deprecated in 2.1, use approx_count_distinct() instead.
approx_count_distinct(col[, rsd])	Aggregate function: returns a new Column for approximate distinct count of column col.
array(*cols)	Creates a new array column.
array_contains(col, value)	Collection function: returns null if the array is null, true if the array contains the given value, and false otherwise.
array_distinct(col)	Collection function: removes duplicate values from the array.
array_except(col1, col2)	Collection function: returns an array of the elements in col1 but not in col2, without duplicates.
array_intersect(col1, col2)	Collection function: returns an array of the elements in the intersection of col1 and col2, without duplicates.
array_join(col, delimiter[, null_replacement])	Concatenates the elements of column using the delimiter.
array_max(col)	Collection function: returns the maximum value of the array.
array_min(col)	Collection function: returns the minimum value of the array.
array_position(col, value)	Collection function: Locates the position of the first occurrence of the given value in the given array.
array_remove(col, element)	Collection function: Remove all elements that equal to element from the given array.
array_repeat(col, count)	Collection function: creates an array containing a column repeated count times.
array_sort(col)	Collection function: sorts the input array in ascending order.
array_union(col1, col2)	Collection function: returns an array of the elements in the union of col1 and col2, without duplicates.
arrays_overlap(a1, a2)	Collection function: returns true if the arrays contain any common non-null element; if not, returns null if both the arrays are non-empty and any of them contains a null element; returns false otherwise.
arrays_zip(*cols)	Collection function: Returns a merged array of structs in which the N-th struct contains all N-th values of input arrays.
asc(col)	Returns a sort expression based on the ascending order of the given column name.
asc_nulls_first(col)	Returns a sort expression based on the ascending order of the given column name, and null values return before non-null values.
asc_nulls_last(col)	Returns a sort expression based on the ascending order of the given column name, and null values appear after non-null values.
ascii(col)	Computes the numeric value of the first character of the string column.
asin(col)	return inverse sine of col, as if computed by java.lang.Math.asin()
atan(col)	return inverse tangent of col, as if computed by java.lang.Math.atan()
atan2(col1, col2)	param col1 coordinate on y-axis
avg(col)	Aggregate function: returns the average of the values in a group.
base64(col)	Computes the BASE64 encoding of a binary column and returns it as a string column.
basestring	alias of builtins.str
bin(col)	Returns the string representation of the binary value of the given column.
bitwiseNOT(col)	Computes bitwise not.
blacklist	Built-in mutable sequence.
broadcast(df)	Marks a DataFrame as small enough for use in broadcast joins.
bround(col[, scale])	Round the given value to scale decimal places using HALF_EVEN rounding mode if scale >= 0 or at integral part when scale < 0.
cbrt(col)	Computes the cube-root of the given value.
ceil(col)	Computes the ceiling of the given value.
coalesce(*cols)	Returns the first column that is not null.
col(col)	Returns a Column based on the given column name.
collect_list(col)	Aggregate function: returns a list of objects with duplicates.
collect_set(col)	Aggregate function: returns a set of objects with duplicate elements eliminated.
column(col)	Returns a Column based on the given column name.
concat(*cols)	Concatenates multiple input columns together into a single column.
concat_ws(sep, *cols)	Concatenates multiple input string columns together into a single string column, using the given separator.
conv(col, fromBase, toBase)	Convert a number in a string column from one base to another.
corr(col1, col2)	Returns a new Column for the Pearson Correlation Coefficient for col1 and col2.
cos(col)	param col angle in radians
cosh(col)	param col hyperbolic angle
count(col)	Aggregate function: returns the number of items in a group.
countDistinct(col, *cols)	Returns a new Column for distinct count of col or cols.
covar_pop(col1, col2)	Returns a new Column for the population covariance of col1 and col2.
covar_samp(col1, col2)	Returns a new Column for the sample covariance of col1 and col2.
crc32(col)	Calculates the cyclic redundancy check value (CRC32) of a binary column and returns the value as a bigint.
create_map(*cols)	Creates a new map column.
cume_dist()	Window function: returns the cumulative distribution of values within a window partition, i.e.
current_date()	Returns the current date as a DateType column.
current_timestamp()	Returns the current timestamp as a TimestampType column.
date_add(start, days)	Returns the date that is days days after start
date_format(date, format)	Converts a date/timestamp/string to a value of string in the format specified by the date format given by the second argument.
date_sub(start, days)	Returns the date that is days days before start
date_trunc(format, timestamp)	Returns timestamp truncated to the unit specified by the format.
datediff(end, start)	Returns the number of days from start to end.
dayofmonth(col)	Extract the day of the month of a given date as integer.
dayofweek(col)	Extract the day of the week of a given date as integer.
dayofyear(col)	Extract the day of the year of a given date as integer.
decode(col, charset)	Computes the first argument into a string from a binary using the provided character set (one of ‘US-ASCII’, ‘ISO-8859-1’, ‘UTF-8’, ‘UTF-16BE’, ‘UTF-16LE’, ‘UTF-16’).
degrees(col)	Converts an angle measured in radians to an approximately equivalent angle measured in degrees.
dense_rank()	Window function: returns the rank of rows within a window partition, without any gaps.
desc(col)	Returns a sort expression based on the descending order of the given column name.
desc_nulls_first(col)	Returns a sort expression based on the descending order of the given column name, and null values appear before non-null values.
desc_nulls_last(col)	Returns a sort expression based on the descending order of the given column name, and null values appear after non-null values
element_at(col, extraction)	Collection function: Returns element of array at given index in extraction if col is array.
encode(col, charset)	Computes the first argument into a binary from a string using the provided character set (one of ‘US-ASCII’, ‘ISO-8859-1’, ‘UTF-8’, ‘UTF-16BE’, ‘UTF-16LE’, ‘UTF-16’).
exists(col, f)	Returns whether a predicate holds for one or more elements in the array.
exp(col)	Computes the exponential of the given value.
explode(col)	Returns a new row for each element in the given array or map.
explode_outer(col)	Returns a new row for each element in the given array or map.
expm1(col)	Computes the exponential of the given value minus one.
expr(str)	Parses the expression string into the column that it represents
factorial(col)	Computes the factorial of the given value.
filter(col, f)	Returns an array of elements for which a predicate holds in a given array.
first(col[, ignorenulls])	Aggregate function: returns the first value in a group.
flatten(col)	Collection function: creates a single array from an array of arrays.
floor(col)	Computes the floor of the given value.
forall(col, f)	Returns whether a predicate holds for every element in the array.
format_number(col, d)	Formats the number X to a format like ‘#,–#,–#.–’, rounded to d decimal places with HALF_EVEN round mode, and returns the result as a string.
format_string(format, *cols)	Formats the arguments in printf-style and returns the result as a string column.
from_csv(col, schema[, options])	Parses a column containing a CSV string to a row with the specified schema.
from_json(col, schema[, options])	Parses a column containing a JSON string into a MapType with StringType as keys type, StructType or ArrayType with the specified schema.
from_unixtime(timestamp[, format])	Converts the number of seconds from unix epoch (1970-01-01 00:00:00 UTC) to a string representing the timestamp of that moment in the current system time zone in the given format.
from_utc_timestamp(timestamp, tz)	This is a common function for databases supporting TIMESTAMP WITHOUT TIMEZONE.
functools	functools.py - Tools for working with functions and callable objects
get_json_object(col, path)	Extracts json object from a json string based on json path specified, and returns json string of the extracted json object.
greatest(*cols)	Returns the greatest value of the list of column names, skipping null values.
grouping(col)	Aggregate function: indicates whether a specified column in a GROUP BY list is aggregated or not, returns 1 for aggregated or 0 for not aggregated in the result set.
grouping_id(*cols)	Aggregate function: returns the level of grouping, equals to
hash(*cols)	Calculates the hash code of given columns, and returns the result as an int column.
hex(col)	Computes hex value of the given column, which could be pyspark.sql.types.StringType, pyspark.sql.types.BinaryType, pyspark.sql.types.IntegerType or pyspark.sql.types.LongType.
hour(col)	Extract the hours of a given date as integer.
hypot(col1, col2)	Computes sqrt(a^2 + b^2) without intermediate overflow or underflow.
ignore_unicode_prefix(f)	Ignore the ‘u’ prefix of string in doc tests, to make it works in both python 2 and 3
initcap(col)	Translate the first letter of each word to upper case in the sentence.
input_file_name()	Creates a string column for the file name of the current Spark task.
instr(str, substr)	Locate the position of the first occurrence of substr column in the given string.
isnan(col)	An expression that returns true iff the column is NaN.
isnull(col)	An expression that returns true iff the column is null.
json_tuple(col, *fields)	Creates a new row for a json column according to the given field names.
kurtosis(col)	Aggregate function: returns the kurtosis of the values in a group.
lag(col[, offset, default])	Window function: returns the value that is offset rows before the current row, and defaultValue if there is less than offset rows before the current row.
last(col[, ignorenulls])	Aggregate function: returns the last value in a group.
last_day(date)	Returns the last day of the month which the given date belongs to.
lead(col[, offset, default])	Window function: returns the value that is offset rows after the current row, and defaultValue if there is less than offset rows after the current row.
least(*cols)	Returns the least value of the list of column names, skipping null values.
length(col)	Computes the character length of string data or number of bytes of binary data.
levenshtein(left, right)	Computes the Levenshtein distance of the two given strings.
lit(col)	Creates a Column of literal value.
locate(substr, str[, pos])	Locate the position of the first occurrence of substr in a string column, after position pos.
log(arg1[, arg2])	Returns the first argument-based logarithm of the second argument.
log10(col)	Computes the logarithm of the given value in Base 10.
log1p(col)	Computes the natural logarithm of the given value plus one.
log2(col)	Returns the base-2 logarithm of the argument.
lower(col)	Converts a string expression to lower case.
lpad(col, len, pad)	Left-pad the string column to width len with pad.
ltrim(col)	Trim the spaces from left end for the specified string value.
map_concat(*cols)	Returns the union of all the given maps.
map_entries(col)	Collection function: Returns an unordered array of all entries in the given map.
map_filter(col, f)	Returns a map whose key-value pairs satisfy a predicate.
map_from_arrays(col1, col2)	Creates a new map from two arrays.
map_from_entries(col)	Collection function: Returns a map created from the given array of entries.
map_keys(col)	Collection function: Returns an unordered array containing the keys of the map.
map_values(col)	Collection function: Returns an unordered array containing the values of the map.
map_zip_with(col1, col2, f)	Merge two given maps, key-wise into a single map using a function.
max(col)	Aggregate function: returns the maximum value of the expression in a group.
md5(col)	Calculates the MD5 digest and returns the value as a 32 character hex string.
mean(col)	Aggregate function: returns the average of the values in a group.
min(col)	Aggregate function: returns the minimum value of the expression in a group.
minute(col)	Extract the minutes of a given date as integer.
monotonically_increasing_id()	A column that generates monotonically increasing 64-bit integers.
month(col)	Extract the month of a given date as integer.
months_between(date1, date2[, roundOff])	Returns number of months between dates date1 and date2.
nanvl(col1, col2)	Returns col1 if it is not NaN, or col2 if col1 is NaN.
next_day(date, dayOfWeek)	Returns the first date which is later than the value of the date column.
ntile(n)	Window function: returns the ntile group id (from 1 to n inclusive) in an ordered window partition.
overlay(src, replace, pos[, len])	Overlay the specified portion of src with replace, starting from byte position pos of src and proceeding for len bytes.
pandas_udf([f, returnType, functionType])	Creates a pandas user defined function (a.k.a.
percent_rank()	Window function: returns the relative rank (i.e.
percentile_approx(col, percentage[, accuracy])	Returns the approximate percentile value of numeric column col at the given percentage.
posexplode(col)	Returns a new row for each element with position in the given array or map.
posexplode_outer(col)	Returns a new row for each element with position in the given array or map.
pow(col1, col2)	Returns the value of the first argument raised to the power of the second argument.
quarter(col)	Extract the quarter of a given date as integer.
radians(col)	Converts an angle measured in degrees to an approximately equivalent angle measured in radians.
rand([seed])	Generates a random column with independent and identically distributed (i.i.d.) samples uniformly distributed in [0.0, 1.0).
randn([seed])	Generates a column with independent and identically distributed (i.i.d.) samples from the standard normal distribution.
rank()	Window function: returns the rank of rows within a window partition.
regexp_extract(str, pattern, idx)	Extract a specific group matched by a Java regex, from the specified string column.
regexp_replace(str, pattern, replacement)	Replace all substrings of the specified string value that match regexp with rep.
repeat(col, n)	Repeats a string column n times, and returns it as a new string column.
reverse(col)	Collection function: returns a reversed string or an array with reverse order of elements.
rint(col)	Returns the double value that is closest in value to the argument and is equal to a mathematical integer.
round(col[, scale])	Round the given value to scale decimal places using HALF_UP rounding mode if scale >= 0 or at integral part when scale < 0.
row_number()	Window function: returns a sequential number starting at 1 within a window partition.
rpad(col, len, pad)	Right-pad the string column to width len with pad.
rtrim(col)	Trim the spaces from right end for the specified string value.
schema_of_csv(csv[, options])	Parses a CSV string and infers its schema in DDL format.
schema_of_json(json[, options])	Parses a JSON string and infers its schema in DDL format.
second(col)	Extract the seconds of a given date as integer.
sequence(start, stop[, step])	Generate a sequence of integers from start to stop, incrementing by step.
sha1(col)	Returns the hex string result of SHA-1.
sha2(col, numBits)	Returns the hex string result of SHA-2 family of hash functions (SHA-224, SHA-256, SHA-384, and SHA-512).
shiftLeft(col, numBits)	Shift the given value numBits left.
shiftRight(col, numBits)	(Signed) shift the given value numBits right.
shiftRightUnsigned(col, numBits)	Unsigned shift the given value numBits right.
shuffle(col)	Collection function: Generates a random permutation of the given array.
signum(col)	Computes the signum of the given value.
sin(col)	param col angle in radians
since(version)	A decorator that annotates a function to append the version of Spark the function was added.
sinh(col)	param col hyperbolic angle
size(col)	Collection function: returns the length of the array or map stored in the column.
skewness(col)	Aggregate function: returns the skewness of the values in a group.
slice(x, start, length)	Collection function: returns an array containing all the elements in x from index start (array indices start at 1, or from the end if start is negative) with the specified length.
sort_array(col[, asc])	Collection function: sorts the input array in ascending or descending order according to the natural ordering of the array elements.
soundex(col)	Returns the SoundEx encoding for a string
spark_partition_id()	A column for partition ID.
split(str, pattern[, limit])	Splits str around matches of the given pattern.
sqrt(col)	Computes the square root of the specified float value.
stddev(col)	Aggregate function: alias for stddev_samp.
stddev_pop(col)	Aggregate function: returns population standard deviation of the expression in a group.
stddev_samp(col)	Aggregate function: returns the unbiased sample standard deviation of the expression in a group.
struct(*cols)	Creates a new struct column.
substring(str, pos, len)	Substring starts at pos and is of length len when str is String type or returns the slice of byte array that starts at pos in byte and is of length len when str is Binary type.
substring_index(str, delim, count)	Returns the substring from string str before count occurrences of the delimiter delim.
sum(col)	Aggregate function: returns the sum of all values in the expression.
sumDistinct(col)	Aggregate function: returns the sum of distinct values in the expression.
sys	This module provides access to some objects used or maintained by the interpreter and to functions that interact strongly with the interpreter.
tan(col)	param col angle in radians
tanh(col)	param col hyperbolic angle
toDegrees(col)	Note Deprecated in 2.1, use degrees() instead.
toRadians(col)	Note Deprecated in 2.1, use radians() instead.
to_csv(col[, options])	Converts a column containing a StructType into a CSV string.
to_date(col[, format])	Converts a Column into pyspark.sql.types.DateType using the optionally specified format.
to_json(col[, options])	Converts a column containing a StructType, ArrayType or a MapType into a JSON string.
to_str(value)	A wrapper over str(), but converts bool values to lower case strings.
to_timestamp(col[, format])	Converts a Column into pyspark.sql.types.TimestampType using the optionally specified format.
to_utc_timestamp(timestamp, tz)	This is a common function for databases supporting TIMESTAMP WITHOUT TIMEZONE.
transform(col, f)	Returns an array of elements after applying a transformation to each element in the input array.
transform_keys(col, f)	Applies a function to every key-value pair in a map and returns a map with the results of those applications as the new keys for the pairs.
transform_values(col, f)	Applies a function to every key-value pair in a map and returns a map with the results of those applications as the new values for the pairs.
translate(srcCol, matching, replace)	A function translate any character in the srcCol by a character in matching.
trim(col)	Trim the spaces from both ends for the specified string column.
trunc(date, format)	Returns date truncated to the unit specified by the format.
udf([f, returnType])	Creates a user defined function (UDF).
unbase64(col)	Decodes a BASE64 encoded string column and returns it as a binary column.
unhex(col)	Inverse of hex.
unix_timestamp([timestamp, format])	Convert time string with given pattern (‘yyyy-MM-dd HH:mm:ss’, by default) to Unix time stamp (in seconds), using the default timezone and the default locale, return null if fail.
upper(col)	Converts a string expression to upper case.
var_pop(col)	Aggregate function: returns the population variance of the values in a group.
var_samp(col)	Aggregate function: returns the unbiased sample variance of the values in a group.
variance(col)	Aggregate function: alias for var_samp.
warnings	Python part of the warnings subsystem.
weekofyear(col)	Extract the week number of a given date as integer.
when(condition, value)	Evaluates a list of conditions and returns one of multiple possible result expressions.
window(timeColumn, windowDuration[, …])	Bucketize rows into one or more time windows given a timestamp specifying column.
xxhash64(*cols)	Calculates the hash code of given columns using the 64-bit variant of the xxHash algorithm, and returns the result as a long column.
year(col)	Extract the year of a given date as integer.
zip_with(col1, col2, f)	Merge two given arrays, element-wise, into a single array using a function.

from_avro(data, jsonFormatSchema[, options])	Converts a binary column of Avro format into its corresponding catalyst value.
to_avro(data[, jsonFormatSchema])	Converts a column into binary of avro format.

Window

Window.currentRow
Window.orderBy(*cols)	Creates a WindowSpec with the ordering defined.
Window.partitionBy(*cols)	Creates a WindowSpec with the partitioning defined.
Window.rangeBetween(start, end)	Creates a WindowSpec with the frame boundaries defined, from start (inclusive) to end (inclusive).
Window.rowsBetween(start, end)	Creates a WindowSpec with the frame boundaries defined, from start (inclusive) to end (inclusive).
Window.unboundedFollowing
Window.unboundedPreceding

Grouping

GroupedData.agg(*exprs)	Compute aggregates and returns the result as a DataFrame.
GroupedData.apply(udf)	It is an alias of pyspark.sql.GroupedData.applyInPandas(); however, it takes a pyspark.sql.functions.pandas_udf() whereas pyspark.sql.GroupedData.applyInPandas() takes a Python native function.
GroupedData.applyInPandas(func, schema)	Maps each group of the current DataFrame using a pandas udf and returns the result as a DataFrame.
GroupedData.avg(*cols)	Computes average values for each numeric columns for each group.
GroupedData.cogroup(other)	Cogroups this group with another group so that we can run cogrouped operations.
GroupedData.count()	Counts the number of records for each group.
GroupedData.max(*cols)	Computes the max value for each numeric columns for each group.
GroupedData.mean(*cols)	Computes average values for each numeric columns for each group.
GroupedData.min(*cols)	Computes the min value for each numeric column for each group.
GroupedData.pivot(pivot_col[, values])	Pivots a column of the current DataFrame and perform the specified aggregation.
GroupedData.sum(*cols)	Compute the sum for each numeric columns for each group.

Structured Streaming

Structured Streaming is a scalable and fault-tolerant stream processing engine built on the Spark SQL engine. You can express your streaming computation the same way you would express a batch computation on static data. The Spark SQL engine will take care of running it incrementally and continuously and updating the final result as streaming data continues to arrive. You can use the Dataset/DataFrame to express streaming aggregations, event-time windows, stream-to-batch joins, etc. The computation is executed on the same optimized Spark SQL engine. Finally, the system ensures end-to-end exactly-once fault-tolerance guarantees through checkpointing and Write-Ahead Logs. In short, Structured Streaming provides fast, scalable, fault-tolerant, end-to-end exactly-once stream processing without the user having to reason about streaming.

Note that you can leverage almost all Spark SQL APIs in PySpark. This page describes Structured Streaming specific features.

Core Classes

The classes below are core classes in Structured Streaming APIs at PySpark.

DataStreamReader(spark)	Interface used to load a streaming DataFrame from external storage systems (e.g.
DataStreamWriter(df)	Interface used to write a streaming DataFrame to external storage systems (e.g.
ForeachBatchFunction(sql_ctx, func)	This is the Python implementation of Java interface ‘ForeachBatchFunction’.
StreamingQuery(jsq)	A handle to a query that is executing continuously in the background as new data arrives.
StreamingQueryException(desc, stackTrace[, …])	Exception that stopped a StreamingQuery.
StreamingQueryManager(jsqm)	A class to manage all the StreamingQuery StreamingQueries active.

Input and Output

DataStreamReader.csv(path[, schema, sep, …])	Loads a CSV file stream and returns the result as a DataFrame.
DataStreamReader.format(source)	Specifies the input data source format.
DataStreamReader.json(path[, schema, …])	Loads a JSON file stream and returns the results as a DataFrame.
DataStreamReader.load([path, format, schema])	Loads a data stream from a data source and returns it as a DataFrame.
DataStreamReader.option(key, value)	Adds an input option for the underlying data source.
DataStreamReader.options(**options)	Adds input options for the underlying data source.
DataStreamReader.orc(path[, mergeSchema, …])	Loads a ORC file stream, returning the result as a DataFrame.
DataStreamReader.parquet(path[, …])	Loads a Parquet file stream, returning the result as a DataFrame.
DataStreamReader.schema(schema)	Specifies the input schema.
DataStreamReader.text(path[, wholetext, …])	Loads a text file stream and returns a DataFrame whose schema starts with a string column named “value”, and followed by partitioned columns if there are any.
DataStreamWriter.foreach(f)	Sets the output of the streaming query to be processed using the provided writer f.
DataStreamWriter.foreachBatch(func)	Sets the output of the streaming query to be processed using the provided function.
DataStreamWriter.format(source)	Specifies the underlying output data source.
DataStreamWriter.option(key, value)	Adds an output option for the underlying data source.
DataStreamWriter.options(**options)	Adds output options for the underlying data source.
DataStreamWriter.outputMode(outputMode)	Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
DataStreamWriter.partitionBy(*cols)	Partitions the output by the given columns on the file system.
DataStreamWriter.queryName(queryName)	Specifies the name of the StreamingQuery that can be started with start().
DataStreamWriter.start([path, format, …])	Streams the contents of the DataFrame to a data source.
DataStreamWriter.trigger([processingTime, …])	Set the trigger for the stream query.

Query Management

StreamingQuery.awaitTermination([timeout])	Waits for the termination of this query, either by query.stop() or by an exception.
StreamingQuery.exception()	return the StreamingQueryException if the query was terminated by an exception, or None.
StreamingQuery.explain([extended])	Prints the (logical and physical) plans to the console for debugging purpose.
StreamingQuery.id	Returns the unique id of this query that persists across restarts from checkpoint data.
StreamingQuery.isActive	Whether this streaming query is currently active or not.
StreamingQuery.lastProgress	Returns the most recent StreamingQueryProgress update of this streaming query or None if there were no progress updates
StreamingQuery.name	Returns the user-specified name of the query, or null if not specified.
StreamingQuery.processAllAvailable()	Blocks until all available data in the source has been processed and committed to the sink.
StreamingQuery.recentProgress	Returns an array of the most recent [[StreamingQueryProgress]] updates for this query.
StreamingQuery.runId	Returns the unique id of this query that does not persist across restarts.
StreamingQuery.status	Returns the current status of the query.
StreamingQuery.stop()	Stop this streaming query.
StreamingQueryManager.active	Returns a list of active queries associated with this SQLContext
StreamingQueryManager.awaitAnyTermination([…])	Wait until any of the queries on the associated SQLContext has terminated since the creation of the context, or since resetTerminated() was called.
StreamingQueryManager.get(id)	Returns an active query from this SQLContext or throws exception if an active query with this name doesn’t exist.
StreamingQueryManager.resetTerminated()	Forget about past terminated queries so that awaitAnyTermination() can be used again to wait for new terminations.

ML

Classification

Clustering

Recommendation

Reression

Statistics

Turning

Evaluation

Image

Spark Streaming

Core Classes

Input and Output

Streaming Context

DStream

Kinesis

MLlib

Classification

Clustering

Recommendation

Reression

Statistics

Turning

Evaluation

Image

Spark Core

Core Classes

Spark Context APIs

Input and Output

RDD APIs

Resource Management

Resource Allocation

Resource Profile

Development

Developer Tools

Testing PySpark

To run individual PySpark tests, you can use run-tests script under python directory. Test cases are located at tests package under each PySpark packages. Note that, if you add some changes into Scala or Python side in Apache Spark, you need to manually build Apache Spark again before running PySpark tests in order to apply the changes. Running PySpark testing script does not automatically build it.

Also, note that there is an ongoing issue to use PySpark on macOS High Serria+. OBJC_DISABLE_INITIALIZE_FORK_SAFETY should be set to YES in order to run some of tests. See PySpark issue and Python issue for more details.

To run test cases in a specific module:

$ python/run-tests --testnames pyspark.sql.tests.test_arrow

To run test cases in a specific class:

$ python/run-tests --testnames 'pyspark.sql.tests.test_arrow ArrowTests'

To run single test case in a specific class:

$ python/run-tests --testnames 'pyspark.sql.tests.test_arrow ArrowTests.test_null_conversion'

You can also run doctests in a specific module:

$ python/run-tests --testnames pyspark.sql.dataframe

Lastly, there is another script called run-tests-with-coverage in the same location, which generates coverage report for PySpark tests. It accepts same arguments with run-tests.

$ python/run-tests-with-coverage --testnames pyspark.sql.tests.test_arrow --python-executables=python
...
Name                              Stmts   Miss Branch BrPart  Cover
-------------------------------------------------------------------
pyspark/__init__.py                  42      4      8      2    84%
pyspark/_globals.py                  16      3      4      2    75%
...

Generating HTML files for PySpark coverage under /…/spark/python/test_coverage/htmlcov You can check the coverage report visually by HTMLs under /…/spark/python/test_coverage/htmlcov.

Please check other available options via python/run-tests[-with-coverage] --help.

Setup Pycharm with PySpark

Import a project to PyCharm: File –> Open –> path_to_project.

After that, open ~/.bash_profile and add the below lines.

# Spark home
export SPARK_HOME=/.../spark-2.4.5-bin-hadoop2.7
export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin
export PYTHONPATH=$SPARK_HOME/python/:$PYTHONPATH
export PYTHONPATH=$SPARK_HOME/python/lib/py4j-0.10.4-src.zip:$PYTHONPATH

Source the ~/.bash_profile to reflect the changes.

source ~/.bash_profile

Install pyspark and pypandoc: PyCharm –> Preferences –> Project Interpreter

Go to PyCharm –> Preferences –> Project Interpreter. Click on Add Content Root. Here you need to add paths

Restart PyCharm, and then run the project. You should be able to see output.

+---------+-----+
|     word|count|
+---------+-----+
|      ...|    5|
|     July|    2|
|       By|    1|
|   North,|    1|
|   taking|    1|
|    harry|   18|
|     #TBT|    1|
|  Potter:|    3|
|character|    1|
|        7|    2|
|  Phoenix|    1|
|   Number|    1|
|      day|    1|
|   (Video|    1|
| seconds)|    1|
| Hermione|    3|
|    Which|    1|
|      did|    1|
|   Potter|   38|
|Voldemort|    1|
+---------+-----+
only showing top 20 rows

Process finished with exit code 0

Contribution Guide

Style Guide

Filing an Issue

Review Process

Release Process

setup.py

Official Release Channel

PyPi and PIP

Roadmaps

Apache Spark 3.0

Pandas UDF and Pandas Function APIs

Extendability

Release Notes

Apache Spark 3.0

Apache Spark 2.4.6

Apache Spark 2.4.5

Spark News

Spark 2.4.5 released

We are happy to announce the availability of Spark 2.4.5! Visit the release notes to read about the new features, or download the release today.

Preview release of Spark 3.0

To enable wide-scale community testing of the upcoming Spark 3.0 release, the Apache Spark community has posted a Spark 3.0.0 preview2 release. This preview is not a stable release in terms of either API or functionality, but it is meant to give the community early access to try the code that will become Spark 3.0. If you would like to test the release, please download it, and send feedback using either the mailing lists or JIRA. The documentation is available at the link.