# -*- coding: utf-8 -*-

# PySpark RDD Operations Example

# Usage:

# spark-submit rdd_operations.py <input_file> <output_folder>

from pyspark import SparkContext, SparkConf

import sys

# =============================

# STEP 1: ARGUMENT HANDLING

# =============================

if len([Link]) != 3:

print("Usage: spark-submit rdd_operations.py <input_file> <output_folder>")

[Link](1)

input_path = [Link][1]

output_path = [Link][2]

# =============================

# STEP 2: SPARK CONTEXT

# =============================

conf = SparkConf().setAppName("PySpark RDD Operations").setMaster("local[*]")

sc = SparkContext(conf=conf)
# =============================

# STEP 3: CREATION OPERATIONS

# =============================

# 1. Parallelize

data = [1, 2, 3, 4, 5, 6, 7, 8, 9]

rdd = [Link](data)

print("Parallelized RDD:", [Link]())

# 2. TextFile (input)

text_rdd = [Link](input_path)

print("Text File RDD Sample:", text_rdd.take(5))

# =============================

# STEP 4: BASIC TRANSFORMATIONS

# =============================

# Map

map_rdd = [Link](lambda x: x * 2)

# FlatMap

flatmap_rdd = text_rdd.flatMap(lambda line: [Link](" "))

# Filter

filter_rdd = [Link](lambda x: x % 2 == 0)
# Set Operations

rdd1 = [Link]([1, 2, 3, 4])

rdd2 = [Link]([3, 4, 5, 6])

union_rdd = [Link](rdd2)

intersection_rdd = [Link](rdd2)

subtract_rdd = [Link](rdd2)

# SortBy

sortby_rdd = [Link](lambda x: -x)

# =============================

# STEP 5: BASIC ACTIONS

# =============================

collect_res = [Link]()

take_res = [Link](3)

top_res = [Link](3)

count_res = [Link]()

reduce_res = [Link](lambda a, b: a + b)

fold_res = [Link](0, lambda a, b: a + b)

aggregate_res = [Link]((0, 0),

(lambda acc, value: (acc[0] + value, acc[1] + 1)),

(lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])))

# =============================

# STEP 6: KEY-VALUE TRANSFORMATIONS

# Create Key-Value Pair RDD

pairs = [Link]([("a", 1), ("b", 2), ("a", 3), ("b", 4), ("c", 5)])

# ReduceByKey

reducebykey_rdd = [Link](lambda x, y: x + y)

# GroupByKey

groupbykey_rdd = [Link]().mapValues(list)

# CombineByKey

combinebykey_rdd = [Link](

lambda value: (value, 1),

lambda acc, value: (acc[0] + value, acc[1] + 1),

lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])

# SortByKey

sortbykey_rdd = [Link]()

# SubtractByKey

pairs2 = [Link]([("a", 100), ("c", 200)])

subtractbykey_rdd = [Link](pairs2)

# MapValues and FlatMapValues

flatmapvalues_rdd = [Link](lambda x: range(x))

# Keys and Values

keys_rdd = [Link]()

values_rdd = [Link]()

# =============================

# STEP 7: KEY-VALUE ACTIONS

# =============================

# Items (collect as pairs)

items_res = [Link]()

# Lookup

lookup_res = [Link]("a")

# =============================

# STEP 8: SAVE RESULTS TO TEXT FILE

# =============================

# Combine some results for saving

final_output = [

"=== BASIC ACTIONS ===",

f"Collect: {collect_res}",

f"Take: {take_res}",
 f"Top: {top_res}",

f"Count: {count_res}",

f"Reduce: {reduce_res}",

f"Fold: {fold_res}",

f"Aggregate: Sum={aggregate_res[0]}, Count={aggregate_res[1]}",

"",

"=== KEY-VALUE OPERATIONS ===",

f"ReduceByKey: {reducebykey_rdd.collect()}",

f"GroupByKey: {groupbykey_rdd.collect()}",

f"CombineByKey (sum,count): {combinebykey_rdd.collect()}",

f"SortByKey: {sortbykey_rdd.collect()}",

f"SubtractByKey: {subtractbykey_rdd.collect()}",

f"MapValues: {mapvalues_rdd.collect()}",

f"FlatMapValues: {flatmapvalues_rdd.collect()}",

f"Keys: {keys_rdd.collect()}",

f"Values: {values_rdd.collect()}",

f"Lookup('a'): {lookup_res}"

# Save as text file

[Link](final_output).saveAsTextFile(output_path)

# =============================

# STEP 9: STOP SPARK

# =============================

[Link]()