Gain insights into enhancing Python data processing with PySpark. Delve into reading, transforming, aggregating data, and creating user-defined functions, boosting efficiency with Apache Spark.

frompandastopyspark.tar.gz

PandasAndPyspark

jupyter

spa-jupyter

spa-snapshot

spa-copy

spa-copy-copy

spa-copy-copy-5klyxp

spa-snapshot-copy

spa-snapshot-copy-copy

spa-copy-copy-5klyxp-copy

Pandas is a popular Python library used to manipulate data, but it has certain limitations in its ability to process large datasets. The Apache Spark analytics library offers significant performance improvements.

This course will help improve your Python-based data processing by leveraging Apache Spark’s multithreading capabilities through the PySpark library. You’ll start by reading data into a PySpark DataFrame before performing basic input/output functions, such as renaming attributes, selecting, and writing data. You’ll move onto transformation functions like aggregation, statistical analysis, and joins before creating custom, user-defined functions. At each step, you’ll get a quick Pandas review before being walked through leveraging the more robust PySpark library to unlock Apache Spark.

By the end of this course, you’ll be able to quickly and reliably process large amounts of data, even stored across multiple files, using PySpark.

From Pandas to PySpark DataFrame

# Create an in-memory DataFrame
To elaborate a more complex application of UDF, let’s first create an in-memory DataFrame as follows:

import pandas as pd
PANDAS_DISPLAY_SETTING = [
    ('display.max_rows', 500),
    ('display.max_columns', 500),
    ('display.width', 1000)
]


def set_pd_display(module: pd) -> bool:
    for item in PANDAS_DISPLAY_SETTING:
        module.set_option(*item)
    return True


# pathconfig.py
from pathlib import Path
import time
import yaml

ROOT = Path(__file__).parent
DATA_LAKE = ROOT.joinpath("data")
RAW_DATA = DATA_LAKE.joinpath("raw")
SNAPSHOT = DATA_LAKE.joinpath("snapshot")

PATH_TOY_GAME_DATA = RAW_DATA.joinpath("Toys_and_Games_5.json")
PATH_HOME_KITCHEN_DATA = RAW_DATA.joinpath("sample_Home_and_Kitchen_5.json")
PATH_HOME_KITCHEN_METADATA = RAW_DATA.joinpath("sample_meta_Home_and_Kitchen.json")


def create_path_snapshot() -> Path:
    """
    Create path to shapshot wit time stamp
    :return: Path like object
    """
    path_fixed = 'snapshot_{}'.format(int(time.time()))
    return SNAPSHOT.joinpath(path_fixed)


# TODO: Store The Snapshot
def create_path_snapshot_spark():
    path_fixed = 'data/snapshot/pyspark/snapshot_{}'
    current_unix_time = int(time.time())
    return path_fixed.format(current_unix_time)


def load_latest_parquet_path() -> Path:
    """
    :return: path to latest parquet file storage
    """
    with open(SNAPSHOT.joinpath('pyspark', 'versions.yaml')) as f:
        content = yaml.safe_load(f)
        print(content)
        latest = content['latest']
        path = content[latest]['path']
        return SNAPSHOT.joinpath(path)



from utils.session import create_spark_session
from pyspark.sql.types import FloatType, IntegerType, ArrayType,  StringType,  StructType,  StructField,  BooleanType
import pyspark.sql.functions as fn
from pyspark.sql.functions import col
import re

spark = create_spark_session()


# Example 1: Calculate Sample Mean
def get_sample_mean(data: list):
    return sum(data)/(len(data) - 1)


udf_get_sample_mean = fn.udf(
    f=get_sample_mean,
    returnType=FloatType()
)


# Example 2
def split_text(data: str):
    names = data.replace("\n",  "").split()
    if len(names) > 2:
        return [names[0],  names[-1]]
    return names


udf_split_text = fn.udf(
    f=split_text,
    returnType=ArrayType(StringType(),  True)
)


data = [
    ("James", "", "Smith", "36636", "M", "Down Hill, 123456, CA"),
    ("Michael", "Rose", "", "40288", "M", "Up Hill, 123457, CA"),
    ("Robert", "", "Williams", "42114", "M", "Central Hill, 123458, CA"),
    ("Maria", "Anne", "Jones", "39192", "F", "Down Town, 123459, CA"),
    ("Jen", "Mary", "Brown", "", "F", "West Wild, 123452, CA")
]

df = spark.createDataFrame(
    data=data,
    schema=["firstname", "middlename", "lastname", "id", "gender", "address"]
)

df.show()

print('Code Executed Successfully')



React

Dart

GoJS React

Typescript React

Vue.js

After successful code execution, we’ll see the message “Code Executed Successfully” in the terminal.

# Adding a new column to DataFrame

Let’s say we want to add a new column that summarizes the `Address` column of the DataFrame. The new `summary` column should have three fields— the length of the full address, a boolean that indicates if there is a postcode or not, and the postcode itself. We can write the UDF as follows:

# Alternate way of defining Specific Schema
schema = StructType([
    StructField("firstname", StringType(), True),
    StructField("middlename", StringType(), True),
    StructField("lastname", StringType(), True),
    StructField("id", StringType(), True),
    StructField("gender", StringType(), True),
    StructField("address", StringType(), True),
])

df = spark.createDataFrame(
    data=data,
    schema=schema
)


# Example 3
def make_summary(text: str) -> dict:
    """
    Summarize an address
    :param text: An address
    :return: Dictionary of Information
    """
    has_number = False
    zipcode = ""
    len_text = len(text)
    numbers = re.search(r'\d+', text)
    if numbers:
        has_number = True
        zipcode = numbers.group()
    return {"len": len_text,  "num_present": has_number, "zipcode": zipcode}


udf_make_summery = fn.udf(
    f=make_summary, 
    returnType=StructType([
        StructField('len',  IntegerType(),  False),
        StructField('num_present',  BooleanType()),
        StructField('zipcode', StringType())
    ])
)

df_with_summary = df.withColumn("summary", udf_make_summery("address"))
df_with_summary.select(col("summary.len").alias("address_length")).show()
df_with_summary.select("address", "summary").show()
df_with_summary.select("id", "summary.zipcode", "summary.num_present").show()

print('Code Executed Successfully')




After successful code execution, we’ll see the message “Code Executed Successfully” in the terminal.

We first define a Python function, `make_summary`. This function returns a dictionary with three key-value pairs. Then we wrap this function with the `fn.udf` function and a proper return type. Here, the return type is more complex—the Python dictionary is mirrored with a list of `StructType(`) and each key-value pair is mirrored with `StructField()`. The first argument of `StructField` is the field name (key) and the second argument is the type of the field (type of the value). We can also provide an optional argument to determine if the field could be `Null` or not.

Learn to play around with user-defined functions.


Introduction

Data Input/Output

Data Transformation

User Defined Function (UDF)

Wrapping Up

Appendix

UDF in Action

Create an in-memory DataFrame

Adding a new column to DataFrame