<jupyter_start><jupyter_text>CREATING RDDs USING PYTHON COLLECTIONS<jupyter_code>sc.parallelize(['this', 'is', 'an', 'example'])
range_rdd = sc.parallelize(range(1,100))
range_rdd.collect()
# range_rdd.count()<jupyter_output><empty_output><jupyter_text>CREATING RDDs USING TEXT FILES<jupyter_code>rdd_hp = sc.textFile('data/HP.txt')
rdd_hp # Check the type of the RDD
# Check the contents of the rdd created
rdd_hp.collect()
# Check the total elements in the RDD<jupyter_output><empty_output><jupyter_text>SPARK TRANSFORMATIONS & ACTIONS  map(func) : Return a new distributed dataset formed by passing each element of the source through a function func.<jupyter_code>ten_range_rdd = sc.parallelize(range(1,10)) # [1,2,.....10]
ten_range_rdd_mapped = ten_range_rdd.map(lambda x: (x, x)) # RDD created from an RDD
ten_range_rdd_mapped.collect()  # collect is an action!
ten_range_rdd_mapped_again = ten_range_rdd_mapped.map(lambda x: (x[0]*2, x[1])).collect() # Python collection
print(ten_range_rdd_mapped_again)<jupyter_output><empty_output><jupyter_text>flatMap(func) : Similar to map, but flattens the final result.<jupyter_code>ten_range_rdd_flatmapped = ten_range_rdd_mapped.flatMap(lambda x: (x[0], x[1] + 1))
ten_range_rdd_flatmapped.collect()<jupyter_output><empty_output><jupyter_text>filter(func) : return a new dataset formed by selecting those elements of the source on which func returns true.<jupyter_code>rdd_hp.collect() # Get back to the RDD we created
rdd_hp_filtered = rdd_hp.filter(lambda line: line.startswith('If'))
rdd_hp_filtered.collect()<jupyter_output><empty_output><jupyter_text>USER EXERCISE<jupyter_code># Convert all the lines to Uppercase
# Remove the lines which contain the word "HIDDEN” in it
# Create a new RDD with elements that are tuples (x, y), where x represents a line and y is the length of that line
# Remove lines that are longer than 30 characters
# Arrange all the words in a single list and count the total number of words
# Arrange the words with the length of each word in a tuple for eg. ("word", 4)<jupyter_output><empty_output><jupyter_text>groupByKey() : When called on a dataset of (K, V) pairs, returns a dataset of (K, Iterable &lt;V&gt;) pairs.<jupyter_code>rdd_places = sc.parallelize([("Finland", "Helsinki"), ("Norway", "Oslo"), ("Sweden", "Stockholm"),
                             ("Denmark", "Copenhagen"), ("Norway", "Bergen"), ("Finland", "Tampere"),
                             ("Denmark", "Aarhus"), ("Finland", "Turku")])
rdd_places.collect()
rdd_places_grouped = rdd_places.groupByKey()
rdd_places_grouped.collect()
# User readable format<jupyter_output><empty_output><jupyter_text>join(otherDataset, [numTasks]) : When called on datasets of type (K, V) and (K, W), returns a dataset of (K, (V, W)) pairs with all pairs of elements for each key. Outer joins are also supported<jupyter_code>rdd_places_other = sc.parallelize([("Finland", "Espoo"), ("Norway", "Stavanger"), ("Sweden", "Gothenburg"), ("Finland", "Vantaa")])
rdd_places_other.collect()
rdd_places.join(rdd_places_other).collect()
rdd_places_grouped.mapValues(list).join(rdd_places_other).collect()<jupyter_output><empty_output><jupyter_text>cogroup(otherDataset, [numTasks]) : When called on datasets of type (K, V) and (K, W), returns a dataset of (K, (Iterable&lt;V&gt; , Iterable&lt;W&gt;)) tuples.<jupyter_code>rdd_places_other_cogrouped = rdd_places.cogroup(rdd_places_other)
rdd_places_other_cogrouped.collect()
rdd_places_other_cogrouped.map(lambda x: (x[0], (list(x[1][0]), list(x[1][1])))).collect()<jupyter_output><empty_output><jupyter_text>sortByKey([ascending], [numTasks]) : When called on a dataset of (K, V) pairs, returns a dataset of (K, V) pairs sorted by keys in ascending or descending order, as specified in the boolean ascending argument.<jupyter_code># Sort by key
rdd_places.sortByKey().collect()<jupyter_output><empty_output><jupyter_text>reduceByKey(func, [numTasks]) 	When called on a dataset of (K, V) pairs, returns a dataset of (K, V) pairs where the values for each key are aggregated using the given reduce function func, which must be of type (V,V) => V.<jupyter_code>rdd_reduce_example = sc.parallelize([("Messi", 2), ("Suarez", 2), ("Ronaldo", 0), ("Ronaldo", 3), ("Messi", 2)])
rdd_reduce_example.reduceByKey(lambda a,b: a+b).collect()
# Sort by value<jupyter_output><empty_output><jupyter_text>SOME RDD ACTIONS<jupyter_code>rdd_places.collect()
# Check the total elements in rdd_places
# Output the first element in the rdd
# Output an array with the first two elements
# Output a sample of 2 random elements with replacement
# Output the first 2 elements by using their natural ordering
# Output the total number of elements for each key<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12365477948934374, 'fraction_numerical': 0.006752479426039249, 'mean_word_length': 6.6575121163166395, 'pattern_counts': {'":': 0, '<': 49, '<?xml version=': 0, '>': 50, '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': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '403187', 'license': 'MIT', 'n_tokens_mistral': 1696, 'n_tokens_neox': 1636, 'n_words': 459, 'path': 'day1/Apache-Spark-Getting-Started.ipynb', 'repo_name': 'csc-training/spark', 'size': '4739'}