Introduction
DataFrame transformations are operations that allow us to manipulate, filter, aggregate, and modify the data in PySpark. Like RDD transformations, DataFrame transformations are lazily evaluated, meaning they compute only after an action is called.
Below are some of the most used DataFrame transformations in PySpark.
Narrow Transformations
A narrow transformation in PySpark refers to an operation on a DataFrame where each input partition contributes only to a single output partition. Narrow transformations can be computed independently on each partition of the DataFrame without the need to shuffle or exchange data between partitions.
These transformations are more efficient because they can be executed in parallel on each partition of the data without requiring communication between partitions.
| Transformation | Description |
| --------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| filter() | Filters out rows from the DataFrame based on a condition. Example: df.filter(df['age'] > 21) |
| select() | Selects specific columns from the DataFrame. Example: df.select('name', 'age') |
| withColumn() | Adds a new column or replaces an existing one with the transformed values. Example: df.withColumn('agePlusTwo', df['age'] + 2) |
| drop() | Drops specified columns from the DataFrame. Example: df.drop('age') |
| explode() | Creates new rows for each element in an array or map column. Example: df.select(df.name, explode(df.interests)) |
| sample() | Samples a fraction of rows from the DataFrame. Example: df.sample(fraction=0.1) |
| randomSplit() | Splits the DataFrame into multiple DataFrames randomly. Example: df.randomSplit([0.8, 0.2]) |
| withWatermark() | Specifies the watermark for event-time processing. Usually used in streaming data. Example: df.withWatermark('timestamp', '10 minutes') |
| map() | Applies a function to each row in the DataFrame. Example usage typically includes a conversion to RDD: df.rdd.map(lambda row: row.name) |
| flatMap() | Applies a function that returns an iterator to each row in the DataFrame and flattens the results. Example usage typically involves conversion to RDD: df.rdd.flatMap(lambda row: (row.name, row.age)) |
| mapPartitions() | Applies a function to each partition of the DataFrame. Example usage involves conversion to RDD: df.rdd.mapPartitions(lambda iter: iter) |
Wide Transformations
A wide transformation, also known as a shuffle operation, refers to an operation on a DataFrame where data from multiple partitions needs to be combined, shuffled, or aggregated. These operations involve exchanging data between partitions and can impact performance as they may require network communication and data movement across the cluster.
Wide transformations are necessary when computations depend on data that is distributed across multiple partitions and needs to be aggregated or grouped. These operations often involve a shuffle phase, where data is redistributed across partitions based on keys or grouping criteria.
| Transformation | Description |
| -------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| groupBy() | Groups the DataFrame using specified columns, creating a GroupedData object. This is often followed by aggregation operations. Example: df.groupBy('age') |
| agg() | Performs aggregation operations on grouped data. It allows you to specify multiple aggregation expressions at once. Example: df.groupBy('age').agg({'salary': 'mean'}) |
| rollup() | Creates multi-dimensional rollups for aggregation. It's similar to groupBy but allows you to specify multiple levels of aggregation. Example: df.rollup('department', 'age').count() |
| cube() | Creates multi-dimensional cubes for aggregation. It's similar to rollup but generates all possible combinations of grouping columns. Example: df.cube('department', 'age').count() |
| pivot() | Pivots a column of the DataFrame and performs aggregation based on the unique values in that column. Example: df.groupBy('department').pivot('job').sum('salary') |
| window() | Defines a window specification for use in window functions. Window functions allow you to perform calculations across a set of rows related to the current row. Example: from pyspark.sql import Window; windowSpec = Window.partitionBy('department').orderBy('salary') |
| join() | Performs a join operation between two DataFrames. Example: df1.join(df2, df1.id == df2.id, 'inner') |
| reduceBy() | Unlike RDDs, DataFrames do not have a direct reduceBy method. Typically, this logic is handled using groupBy followed by aggregation methods. |
These are just some of the commonly used transformations in the PySpark DataFrame API. There are more transformations available depending on specific use cases and requirements.
These transformations are performed in a distributed and parallel manner, leveraging the parallel processing capabilities of Spark's underlying execution engine. It's important to note that transformations in Spark are immutable, meaning they do not modify the original DataFrame but instead return a new DataFrame with the transformation applied.