<jupyter_start><jupyter_text>Types of RDDs - Basic RDDs : each element/record is a single object- Pair RDDs  : each element is treated like a key/value pair. Each record is a tuple with 2 object<jupyter_code>import os
#  File paths
trafficPath = os.path.join(os.pardir,'data','Dodgers.data.txt')
gamesPath = os.path.join(os.pardir,'data','Dodgers.events.txt')

#  Load the text files into RDDs
traffic = sc.textFile(trafficPath)
games = sc.textFile(gamesPath)
# preview Dodgers.data.txt
traffic.take(5)
# preview Dodgers.events.txt
games.take(10)<jupyter_output><empty_output><jupyter_text>Creating a Pair RDD - Pair RDDs allow you work with data by inherently treating them as pairs ( dimensions and metric ).<jupyter_code>from datetime import datetime
import csv
try:
    from StringIO import StringIO
except ImportError as e:
    from io import StringIO
# function to parse Dodgers.data.txt 
def parseTraffic(row):
    DATE_FMT = "%m/%d/%Y %H:%M"
    row = row.split(',')

    row[0] = datetime.strptime(row[0],DATE_FMT)
    row[1] = int(row[1])
    return (row[0],row[1])
#  create a pair RDD using map function 

trafficParsed = traffic.map(parseTraffic)
# preview the pair RDD
trafficParsed.take(10)
dailyTrend = trafficParsed.map(lambda x: (x[0].date(),x[1]))
dailyTrend.take(5)<jupyter_output><empty_output><jupyter_text>ReduceByKey Example  - reduceByKey : combines records with the same key in a specified way . It is NOT an action rather it returns another RDD. Hence it is a transformation. It only combines values with the same key.<jupyter_code>dailyTrend = trafficParsed.map(lambda x: (x[0].date(),x[1])).reduceByKey(lambda x,y:x+y)
dailyTrend.take(5)<jupyter_output><empty_output><jupyter_text>Joining RDDs<jupyter_code># function to parse Dodgers.events.txt 
def parseGames(row):
    DATE_FMT = "%m/%d/%y"
    row = row.split(',')
    row[0] = datetime.strptime(row[0],DATE_FMT).date()
    return (row[0],row[4])

gamesParsed = games.map(parseGames)
gamesParsed.take(10)
dailyTrendCombined = dailyTrend.leftOuterJoin(gamesParsed) 
dailyTrendCombined.take(5) # notice that the join happens on the keys ( dimensions) and not the values(metrics)
# Function to check if 
def checkGameDay(row):
    if row[1][1] == None:
        return (row[0],row[1][1],"Regular Day",row[1][0])
    else:
        return (row[0],row[1][1],"Game Day",row[1][0])
dailyTrendbyGames=dailyTrendCombined.map(checkGameDay)
dailyTrendbyGames.take(10)
#sorting by the fourth value in each record
dailyTrendbyGames.sortBy(lambda x: -x[3]).take(10)<jupyter_output><empty_output><jupyter_text>combineByKey Example - combineByKey : combine records with the same key in a specified way . You have very granular control on how the computation should happen.	how it works :	combineByKey requires 3 functions. As an example:		combineByKey(func1,func2,func3)* 	func1 : this is called  the createCombiner Function . It initialises a value when a key is first seen within a partition -> . lambda value:(value,1)*       Func2 : this is called the merge function.   This specifies how values with same key should be combined. --->  Lambda acc, value: (acc[0]+value). Where acc is the previous result * Func3 : this is called the mergeCombiners function . This specifies how the results from each partition should be combined.  —> lambda acc1,acc2: (acc1[0]+acc2[0],acc1[1]+acc2[1])<jupyter_code>dailyTrendbyGames.map(lambda x: (x[2],x[3]))\
                .combineByKey(lambda value:(value,1),\
                lambda acc,value:(acc[0]+value,acc[1]+1),\
                lambda acc1,acc2:(acc1[0]+acc2[0],acc1[1]+acc2[1]))\
                .mapValues(lambda x:x[0]/x[1])\
                .collect()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12239935152661442, 'fraction_numerical': 0.021345582275060795, 'mean_word_length': 4.660550458715596, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '610799', 'license': 'MIT', 'n_tokens_mistral': 1310, 'n_tokens_neox': 1252, 'n_words': 345, 'path': 'notebooks/3. Pair RDDs.ipynb', 'repo_name': 'Kelechukwu/Apache-Spark-Beginner-Course', 'size': '3701'}