Pyspark - How to preprocess Large Scale Data with Python

from pyspark.sql import SparkSession

# Create a SparkSession
# SparkSession builder를 초기화. builder는 SparkSession 객체를 설정하고 생성하는데 사용
# 각 Spark 실행자(executor)가 사용할 수 있는 최대 메모리 양을 2GB로 설정
# reduceByKey나 groupBy 같은 변환 후에 데이터를 어떻게 분할할지를 결정. 이이 값은 셔플 과정에서 생성될 파티션의 수를 의미하며, 각 파티션은 하나의 태스크에 의해 처리
spark = SparkSession.builder \
    .appName("MySparkApplication") \
    .config("spark.executor.memory", "2g") \
    .config("spark.sql.shuffle.partitions", "4") \
    .getOrCreate()

from pyspark.sql import SparkSession

# Create a SparkSession
# "master_node_address"는 MASTER 노드 서버의 실제 IP 주소나 호스트 이름. 7077은 Spark의 기본 포트
# spark.executor.memory와 spark.executor.cores는 각 실행자가 사용할 메모리와 CPU 코어 수를 설정

# SparkSession 생성
spark = SparkSession.builder \
    .appName("Example on Standalone Cluster") \
    .master("spark://master_node_address:7077") \
    .config("spark.executor.memory", "4g") \
    .config("spark.executor.cores", "2") \
    .getOrCreate()

# Set the PySpark environment variables

import os
os.environ['SPARK_HOME'] = '/Users/name/App/Spark'
#  SPARK_HOME 환경 변수를 설정하여, PySpark가 Spark이 설치된 위치를 알 수 있도록 설정
os.environ['PYSPARK_DRIVER_PYTHON'] = 'jupyter'
# PYSPARK_DRIVER_PYTHON 환경 변수는 PySpark 세션을 시작할 때 어떤 Python 인터페이스를 사용할지 지정
os.environ['PYSPARK_DRIVER_PYTHON_OPTS'] = 'lab'
# PySpark가 Jupyter Lab을 실행할 때 필요한 추가 옵션을 제공
os.environ['PYSPARK_PYTHON'] = 'python'
# PYSPARK_PYTHON 환경 변수는 클러스터의 모든 노드에서 PySpark 작업을 처리할 때 사용할 Python 인터프리터를
# 이 설정은 PySpark가 작업을 분산 처리할 때 일관된 Python 환경을 유지하도록 돕습니다.

from pyspark.sql import SparkSession
# Create a SparkSession
spark = SparkSession.builder.appName("RDD-Demo").getOrCreate()

'''
현재 Spark 세션이 어떤 마스터 설정을 사용하고 있는지를 출력
### LOCAL
- local[*]이면 로컬 모드에서 모든 사용 가능한 코어를 사용하고 있다는 것을 의미
### CLUSTER
spark://host:port 형식이면 특정 클러스터 매니저(Standalone, Mesos, YARN 등)에 연결되어 있다는 것을 의미
'''
spark = SparkSession.builder.appName("RDD-Demo").getOrCreate()
print(spark.sparkContext.master)

# local[*]

'''
로컬 환경에서 실행 중인 경우, 스레드의 수(즉, 병렬 처리 가능한 작업의 수)는 다음과 같이 확인
'''
import multiprocessing

num_cores = multiprocessing.cpu_count()
print("Number of available cores:", num_cores)

# Number of available cores: 10

# Data를 가용할 수 있는 core(CPU 코어들을 Worker 노드)중 5를 활용해 partitioning

numbers = [1, 2, 3, 4, 5]
rdd = spark.sparkContext.parallelize(numbers,5)

num_partitions = rdd.getNumPartitions()
print("Number of partitions:", num_partitions)

# glom() 함수를 사용하면 각 파티션의 데이터를 배열로 변환하여 파티션별로 그룹화
partitioned_data = rdd.glom().collect()

# print
print("Data in each partition:")
for i, data in enumerate(partitioned_data):
    print(f"Partition {i}: {data}")


# Data in each partition:
# Partition 0: [1]
# Partition 1: [2]
# Partition 2: [3]
# Partition 3: [4]
# Partition 4: [5]

rdd.collect()
# [1, 2, 3, 4, 5]

from pyspark.sql import SparkSession

# 클러스터 모드에서 Spark 세션 설정
spark = SparkSession.builder \
    .master("spark://master_url:7077") \
    .appName("RDD-Demo") \
    .config("spark.executor.instances", "6") \ # 클러스터 전체에서 워커 노드들에 걸쳐 분배
    .config("spark.executor.memory", "2g") \
    .config("spark.executor.cores", "2") \
    .getOrCreate()


# RDD 생성 및 파티션 설정
numbers = list(range(1, 21))
rdd = spark.sparkContext.parallelize(numbers, 6)  # 명시적으로 파티션 수를 6으로 설정

# 각 파티션의 데이터를 리스트로 변환
partitioned_data = rdd.glom().collect()

# 출력
print("Data in each partition:")
for i, data in enumerate(partitioned_data):
    print(f"Partition {i}: {data}")


'''
(예상 출력 결과)
Data in each partition:
Partition 0: [1, 2, 3, 4]
Partition 1: [5, 6, 7, 8]
Partition 2: [9, 10, 11, 12]
Partition 3: [13, 14, 15, 16]
Partition 4: [17, 18]
Partition 5: [19, 20]
'''

filtered_customers = customers_rdd.filter(lambda x: x['age'] > 30)

### LOCAL 노드 환경 가정

# Create an RDD from a list of tuples
data = [("Alice", 25), ("Bob", 30), ("Charlie", 35), ("Alice", 40)]
rdd = spark.sparkContext.parallelize(data)

# a. Map transformation: Convert name to uppercase
mapped_rdd = rdd.map(lambda x: (x[0].upper(), x[1]))
result = mapped_rdd.collect()
print("rdd with uppercease name: ", result)
#rdd with uppercease name:  [('ALICE', 25), ('BOB', 30), ('CHARLIE', 35), ('ALICE', 40)]

# b. Filter transformation: Filter records where age is greater than 30
filtered_rdd = rdd.filter(lambda x: x[1] > 30)
filtered_rdd.collect()
#[('Charlie', 35), ('Alice', 40)]

# c. ReduceByKey transformation: Calculate the total age for each name
reduced_rdd = rdd.reduceByKey(lambda x, y: x + y)
reduced_rdd.collect()
#[('Alice', 65), ('Bob', 30), ('Charlie', 35)]

# d. SortBy transformation: Sort the RDD by age in descending order
sorted_rdd = rdd.sortBy(lambda x: x[1], ascending=False)
sorted_rdd.collect()
#[('Alice', 40), ('Charlie', 35), ('Bob', 30), ('Alice', 25)]

result = filtered_customers.collect()

### LOCAL 노드 환경 가정

# Create an RDD from a list of tuples
data = [("Alice", 25), ("Bob", 30), ("Charlie", 35), ("Alice", 40)]
rdd = spark.sparkContext.parallelize(data)

# a. Collect action: Retrieve all elements of the RDD
print("All elements of the rdd: ", rdd.collect())
# All elements of the rdd:  [('Alice', 25), ('Bob', 30), ('Charlie', 35), ('Alice', 40)]

# b. Count action: Count the number of elements in the RDD
count = rdd.count()
print("The total number of elements in rdd: ", count)
# The total number of elements in rdd:  4

# c. First action: Retrieve the first element of the RDD
first_element = rdd.first()
print("The first element of the rdd: ", first_element)
# The first element of the rdd:  ('Alice', 25)

# d. Take action: Retrieve the n elements of the RDD
taken_elements = rdd.take(2)
print("The first two elements of the rdd: ", taken_elements)
#The first two elements of the rdd:  [('Alice', 25), ('Bob', 30)]

# e. Foreach action: Print each element of the RDD
rdd.foreach(lambda x: print(x))
# ('Charlie', 35)
# ('Alice', 25)
# ('Bob', 30)
# ('Alice', 40)

# create rdd from text file
rdd_text = spark.sparkContext.textFile("output.txt")
rdd_text.collect()

# ["('Alice', 25)", "('Bob', 30)", "('Charlie', 35)", "('Alice', 40)"]

import pandas as pd

# pandas DataFrame 생성
df_pandas = pd.DataFrame({
    'A': [1, 2, 3],
    'B': [4, 5, 6]
})

# 새 컬럼 추가 (inplace가 필요 없는 연산)
df_pandas['C'] = df_pandas['A'] + df_pandas['B']

# 컬럼 이름 변경 (inplace 옵션 사용)
df_pandas.rename(columns={'A': 'Alpha'}, inplace=True)

# 결과 출력
print(df_pandas)


'''
   Alpha  B  C
0      1  4  5
1      2  5  7
2      3  6  9
'''

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# SparkSession 시작
spark = SparkSession.builder.appName("example").getOrCreate()

# Spark DataFrame 생성
df_spark = spark.createDataFrame([
    (1, 4),
    (2, 5),
    (3, 6)
], ["A", "B"])

# 새 컬럼 추가
new_df_spark = df_spark.withColumn('C', col('A') + col('B'))

# 컬럼 이름 변경
final_df_spark = new_df_spark.withColumnRenamed('A', 'Alpha')

# 결과 출력
final_df_spark.show()

'''
+-----+---+---+
|Alpha|  B|  C|
+-----+---+---+
|    1|  4|  5|
|    2|  5|  7|
|    3|  6|  9|
+-----+---+---+
'''

# Setting Environment ...

from pyspark.sql import SparkSession
from pyspark.sql.functions import desc

# Create a SparkSession
spark = SparkSession.builder.appName("DataFrame-Demo").getOrCreate()

rdd = spark.sparkContext.textFile("./data/data.txt")
result_rdd = rdd.flatMap(lambda line: line.split(" ")) \
                .map(lambda word: (word, 1)) \
                .reduceByKey(lambda a, b: a + b) \
                .sortBy(lambda x: x[1], ascending=False)

df = spark.read.text("./data/data.txt")
result_df = df.selectExpr("explode(split(value, ' ')) as word") \
              .groupBy("word") \
              .count() \
              .orderBy(desc("count"))

### Read CSV with header

# Read CSV file into DataFrame
csv_file_path = "./data/products.csv"
df = spark.read.csv(csv_file_path, header=True)

# Display schema of DataFrame
df.printSchema()

# Display content of DataFrame
df.show(5)


'''
root
 |-- id: string (nullable = true)
 |-- name: string (nullable = true)
 |-- category: string (nullable = true)
 |-- quantity: string (nullable = true)
 |-- price: string (nullable = true)

+---+--------------------+---------------+--------+------+
| id|                name|       category|quantity| price|
+---+--------------------+---------------+--------+------+
|  1|           iPhone 12|    Electronics|      10|899.99|
|  2|     Nike Air Max 90|       Clothing|      25|119.99|
|  3|KitchenAid Stand ...|Home Appliances|       5|299.99|
|  4|    The Great Gatsby|          Books|      50| 12.99|
|  5|L'Oreal Paris Mas...|         Beauty|     100|  9.99|
+---+--------------------+---------------+--------+------+
'''

### Read CSV with an explicit schema definition

# import necessary types
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType
# Define the schema
schema = StructType([
    StructField(name="id", dataType=IntegerType(), nullable=True),
    StructField(name="name", dataType=StringType(), nullable=True),
    StructField(name="category", dataType=StringType(), nullable=True),
    StructField(name="quantity", dataType=IntegerType(), nullable=True),
    StructField(name="price", dataType=DoubleType(), nullable=True)
])
# Read CSV file into DataFrame with schema definition
csv_file_path = "./data/products.csv"
df = spark.read.csv(csv_file_path, header=True, schema=schema)
# Display schema of DataFrame
df.printSchema()

# Display content of DataFrame
df.show(5)

'''
root
 |-- id: integer (nullable = true)
 |-- name: string (nullable = true)
 |-- category: string (nullable = true)
 |-- quantity: integer (nullable = true)
 |-- price: double (nullable = true)

+---+--------------------+---------------+--------+------+
| id|                name|       category|quantity| price|
+---+--------------------+---------------+--------+------+
|  1|           iPhone 12|    Electronics|      10|899.99|
|  2|     Nike Air Max 90|       Clothing|      25|119.99|
|  3|KitchenAid Stand ...|Home Appliances|       5|299.99|
|  4|    The Great Gatsby|          Books|      50| 12.99|
|  5|L'Oreal Paris Mas...|         Beauty|     100|  9.99|
+---+--------------------+---------------+--------+------+
only showing top 5 rows
'''

### Read CSV file into DataFrame with inferSchema
csv_file_path = "./data/products.csv"
df = spark.read.csv(csv_file_path, header=True, inferSchema=True)
# Display schema of DataFrame
df.printSchema()

# Display content of DataFrame
df.show(5)

'''
root
 |-- id: integer (nullable = true)
 |-- name: string (nullable = true)
 |-- category: string (nullable = true)
 |-- quantity: integer (nullable = true)
 |-- price: double (nullable = true)

+---+--------------------+---------------+--------+------+
| id|                name|       category|quantity| price|
+---+--------------------+---------------+--------+------+
|  1|           iPhone 12|    Electronics|      10|899.99|
|  2|     Nike Air Max 90|       Clothing|      25|119.99|
|  3|KitchenAid Stand ...|Home Appliances|       5|299.99|
|  4|    The Great Gatsby|          Books|      50| 12.99|
|  5|L'Oreal Paris Mas...|         Beauty|     100|  9.99|
+---+--------------------+---------------+--------+------+
only showing top 5 rows
'''

from pyspark.sql import SparkSession

# Create a SparkSession
spark = SparkSession.builder.appName("DataFrame-Operations").getOrCreate()

# Load the synthetic data into a DataFrame
data_file_path = "./data/stocks.txt"
df = spark.read.csv(data_file_path, header=True, inferSchema=True)

# Select specific columns
selected_columns = df.select("id", "name", "price")
print("Selected Columns:")
selected_columns.show(10)

'''
Selected Columns:
+---+----------------+-------+
| id|            name|  price|
+---+----------------+-------+
|  1|          iPhone| 899.99|
|  2|         Macbook|1299.99|
|  3|            iPad| 499.99|
|  4|      Samsung TV| 799.99|
|  5|           LG TV| 699.99|
|  6|      Nike Shoes|  99.99|
|  7|    Adidas Shoes|  89.99|
|  8| Sony Headphones| 149.99|
|  9|Beats Headphones| 199.99|
| 10|    Dining Table| 249.99|
+---+----------------+-------+
only showing top 10 rows
'''

# Filter rows based on a condition
filtered_data = df.filter(df.quantity > 20)
print("Filtered Data:", filtered_data.count())
filtered_data.show()

'''
Filtered Data: 12
+---+--------------+-----------+--------+-----+
| id|          name|   category|quantity|price|
+---+--------------+-----------+--------+-----+
|  6|    Nike Shoes|   Clothing|      30|99.99|
|  7|  Adidas Shoes|   Clothing|      25|89.99|
| 12|        Apples|       Food|     100|  0.5|
| 13|       Bananas|       Food|     150| 0.25|
| 14|       Oranges|       Food|     120| 0.75|
| 15|Chicken Breast|       Food|      50| 3.99|
| 16| Salmon Fillet|       Food|      30| 5.99|
| 24|    Laptop Bag|Accessories|      25|29.99|
| 25|      Backpack|Accessories|      30|24.99|
| 28|         Jeans|   Clothing|      30|59.99|
| 29|       T-shirt|   Clothing|      50|14.99|
| 30|      Sneakers|   Clothing|      40|79.99|
+---+--------------+-----------+--------+-----+
'''

# GroupBy and Aggregations
grouped_data = df.groupBy("category").agg({"quantity": "sum", "price": "avg"})
print("Grouped and Aggregated Data:")
grouped_data.show()

'''
Grouped and Aggregated Data:
+-----------+-------------+------------------+
|   category|sum(quantity)|        avg(price)|
+-----------+-------------+------------------+
|       Food|          450|2.2960000000000003|
|     Sports|           35|             34.99|
|Electronics|           98| 586.6566666666665|
|   Clothing|          200|  99.2757142857143|
|  Furniture|           41|            141.99|
|Accessories|           55|             27.49|
+-----------+-------------+------------------+
'''

"""
Run this from project root path

python bin/sparkapp.py dedup_job --config_path=./configs/dedup_job.yaml
"""

import random
from itertools import combinations

import yaml
from pyspark import SparkContext
from pyspark.rdd import RDD

from dps.spark.spark_session import spark_session
from dps.spark.utils.io_utils import read_line, to_json
from dps.spark.prep.dedup_prep import (
    shingle_word,
    generate_minhash,
    jaccard_by_hashvalues,
)


def expand_instances_by_minhash(
    data, expand_size: int, n_gram: int, seed: int = 1, char_level: bool = False
):
    shingles = shingle_word(data["text"], n_gram=n_gram, char_level=char_level)
    minhashes = generate_minhash(shingles, num_perm=expand_size, seed=seed)

    for mh in minhashes.tolist():
        yield (str(mh), [dict(**data, shingles=shingles, hashvalues=minhashes)])


def explore_dedup_instance(hash_groups, threshold: float = 0.8):
    if len(hash_groups) <= 1:
        return

    group_represent_text = hash_groups[0][
        "text"
    ]  # not to remove all text instances in group.
    pairs = combinations(hash_groups, 2)

    for d_1, d_2 in pairs:
        sim_score = jaccard_by_hashvalues(d_1["hashvalues"], d_2["hashvalues"])
        if sim_score >= threshold:
            dedup_text = [d_1["text"], d_2["text"]]
            if group_represent_text in dedup_text:
                yield dedup_text[0] if dedup_text[
                    0
                ] != group_represent_text else dedup_text[1]
            else:
                yield random.choice(dedup_text)


def dedup_job(config_path):
    with open(config_path) as f:
        conf = yaml.load(f, Loader=yaml.FullLoader)

    input_paths = ",".join([f'{conf["base_dir"]}/{t}' for t in conf["targets"]])

    with spark_session(f"") as spark:
        sc: SparkContext = spark.sparkContext

        proc_rdd: RDD = (
            sc.textFile(input_paths)
            .repartition(conf["n_dist"])
            .flatMap(read_line)
            .cache()
        )

        overlap_kv_rdd: RDD = (
            proc_rdd.flatMap(
                lambda x: expand_instances_by_minhash(
                    x,
                    expand_size=conf["num_expand"],
                    n_gram=conf["n_gram"],
                    seed=conf["seed"],
                    char_level=conf["char_level"],
                )
            )
            .reduceByKey(lambda x, y: x + y)
            .flatMap(
                lambda x: explore_dedup_instance(x[1], threshold=conf["sim_threshold"])
            )
            .distinct()
            .map(lambda x: (x, dict(text=x)))
            .cache()
        )

        proc_rdd.map(lambda x: (x["text"], x)).subtractByKey(overlap_kv_rdd).map(
            lambda x: x[1]
        ).repartition(conf["n_output"]).flatMap(to_json).saveAsTextFile(
            conf["output_dir"]
        )

def expand_instances_by_minhash(
    data, expand_size: int, n_gram: int, seed: int = 1, char_level: bool = False
):
    shingles = shingle_word(data["text"], n_gram=n_gram, char_level=char_level)
    minhashes = generate_minhash(shingles, num_perm=expand_size, seed=seed)

    for mh in minhashes.tolist():
        yield (str(mh), [dict(**data, shingles=shingles, hashvalues=minhashes)])

def explore_dedup_instance(hash_groups, threshold: float = 0.8):
    if len(hash_groups) <= 1:
        return

    group_represent_text = hash_groups[0]["text"]
    pairs = combinations(hash_groups, 2)

    for d_1, d_2 in pairs:
        sim_score = jaccard_by_hashvalues(d_1["hashvalues"], d_2["hashvalues"])
        if sim_score >= threshold:
            dedup_text = [d_1["text"], d_2["text"]]
            if group_represent_text in dedup_text:
                yield dedup_text[0] if dedup_text[0] != group_represent_text else dedup_text[1]
            else:
                yield random.choice(dedup_text)

proc_rdd: RDD = (
    sc.textFile(input_paths)
    .repartition(conf["n_dist"])
    .flatMap(read_line)
    .cache()
)

overlap_kv_rdd: RDD = (
    proc_rdd.flatMap(
        lambda x: expand_instances_by_minhash(
            x,
            expand_size=conf["num_expand"],
            n_gram=conf["n_gram"],
            seed=conf["seed"],
            char_level=conf["char_level"],
        )
    )
    .reduceByKey(lambda x, y: x + y)
    .flatMap(
        lambda x: explore_dedup_instance(x[1], threshold=conf["sim_threshold"])
    )
    .distinct()
    .map(lambda x: (x, dict(text=x)))
    .cache()
)

proc_rdd.map(lambda x: (x["text"], x)).subtractByKey(overlap_kv_rdd).map(
    lambda x: x[1]
).repartition(conf["n_output"]).flatMap(to_json).saveAsTextFile(
    conf["output_dir"]
)