<jupyter_start><jupyter_text>Loading data into RDD, DataFrame Most of the tasks in this section will be done in the notebook, some will be done on the cluster, you can find the necessary files and submission scripts here:```bashcd 2_LoadingData```By default, Spark will not override the output folder. For `RDDs`, one can pass a config parameter to alter that `.config("spark.hadoop.validateOutputSpecs", "false")` for `DataFrame`s we will explicitely set the write mode. First, initialize spark session and spark context:<jupyter_code>import pyspark
try:
    sc
except NameError:    
    spark = pyspark.sql.SparkSession.builder.master("local[*]").appName("BD course").config("spark.hadoop.validateOutputSpecs", "false").getOrCreate()
    sc = spark.sparkContext<jupyter_output><empty_output><jupyter_text>Load unstructured dataOne of the first things we need to learn is how to read the data into Spark RDDs and dataframes. Spark provides a battery-included API for reading structured data in most data formats (CSV, JSON, Parquet) as well as unstructured data (plain text files, server logs, etc). Reading plain text`textFile` and `wholeTextFiles` are functions to read in plain unstructured text.1. `textFile` reads data line by line creating an RDD where each entry corresponds to a line (kind of like readlines() in Python)1. `wholeTextFiles` reads the whole file into a pair RDD: (file path, context of the whole file as string)Following code demonstrate that on an example of the word count.<jupyter_code>#from pyspark import SparkContext
import sys
import time
import os

def main1(args):
    start = time.time()
    #sc = SparkContext(appName="LoadUnstructured")

    #By default it assumes file located on hdfs folder, 
    #but by prefixing "file://" it will search the local file system
    #Can specify a folder, can pass list of folders or use wild character
    input_rdd = sc.textFile("./data/unstructured/")

    counts = input_rdd.flatMap(lambda line: line.split()).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)

    print ("\nTaking the 10 most frequent words in the text and corresponding frequencies:")
    print (counts.takeOrdered(10, key=lambda x: -x[1]))

    counts.map(lambda x: (x[1],x[0])).sortByKey(0).map(lambda x: (x[1],x[0])).repartition(1).saveAsTextFile("./output_loadunstructured1/")
 
    end = time.time()
    print ("Elapsed time: ", (end-start))
# Try the record-per-line-input
main1(sys.argv)
def main2(args):
    start = time.time()

    #Use alternative approach: load the dinitial file into a pair RDD
    input_pair_rdd = sc.wholeTextFiles("./data/unstructured/")

    counts = input_pair_rdd.flatMap(lambda line: line[1].split()).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)

    print ("\nTaking the 10 most frequent words in the text and corresponding frequencies:")
    print (counts.takeOrdered(10, key=lambda x: -x[1]))
    counts.map(lambda x: (x[1],x[0])).sortByKey(0).map(lambda x: (x[1],x[0])).repartition(1).saveAsTextFile("./output_loadunstructured2/")

    end = time.time()
    print ("Elapsed time: ", (end-start))
#Use alternative approach: load the initial file into a pair RDD
main2(sys.argv)<jupyter_output>Taking the 10 most frequent words in the text and corresponding frequencies:
[('the', 22635), ('of', 11167), ('and', 11086), ('to', 10707), ('a', 10433), ('I', 10183), ('in', 7006), ('that', 6911), ('was', 6779), ('his', 4955)]
Elapsed time:  1.3675246238708496<jupyter_text>Loading CSVNext, we are going to learn how to load data into structured format like CSV. There is at least two ways to do that:1. Read the files line by line with `textFiles()` method, split on delimiter (not recommended). This will produced an RDD which is a data structure optimized for row-oriented analysis and functional primitives like `map` and `filter`1. Read the CSV files using the built in `DataFrameReader` (recommended). This will produce a dataframe, which is a data structure optimized for column-oriented analysis and relational primitives<jupyter_code>import csv
import sys
import os
try:
    from StringIO import StringIO
except ImportError:
    from io import StringIO

#this one is use when you use textFile
def loadRecord(line,header,delimiter):
    """Parse a CSV line"""
    input = StringIO(line)
    reader = csv.DictReader(input, delimiter=delimiter, fieldnames=header)
    return next(reader)

def main_rdd(args):
    #sc = SparkContext(appName="LoadCsv")
    delimiter = "|"

    # Try the record-per-line-input
    input = sc.textFile("./data/csv/person_nodes.csv")
    header = input.first().split(delimiter)
    data = input.filter(lambda x: header[0] not in x).map(lambda x: loadRecord(x,header,delimiter))
    data.repartition(1).saveAsTextFile("./output_csv/")

def main_dataframe(args):
    delimiter = "|"

    #csv into spark dataframe   
    input_df = spark.read.options(header='true', inferschema='true',delimiter=delimiter).csv('./data/csv/person_nodes.csv')
    input_df.write.mode("overwrite").option("header", "true").csv("./output_csv2/")

#Load into a regular RDD using textFile and parsing the CSV file line by line
main_rdd(sys.argv)
#Load into dataframe using the csv reader from Databricks
main_dataframe(sys.argv)<jupyter_output><empty_output><jupyter_text>Example: analyzing a diamonds dataset In a simialr way as before, we are going to read the CSV file into dataframe.Spark DataFrameReader can handle delimiters, escaping, and can optionally skip header line for CSV files.<jupyter_code># Read csv data as DataFrame using spark csv dataframe reader
diamonds = spark.read.options(header='true', inferSchema='true').csv('./data/csv/diamonds.csv')
diamonds.show(10)
diamonds.printSchema()<jupyter_output>root
 |-- _c0: integer (nullable = true)
 |-- carat: double (nullable = true)
 |-- cut: string (nullable = true)
 |-- color: string (nullable = true)
 |-- clarity: string (nullable = true)
 |-- depth: double (nullable = true)
 |-- table: double (nullable = true)
 |-- price: integer (nullable = true)
 |-- x: double (nullable = true)
 |-- y: double (nullable = true)
 |-- z: double (nullable = true)<jupyter_text>Let's try doing some basic queries to understand the dataset better.<jupyter_code>diamonds.count()
diamonds.select('color').distinct().show()<jupyter_output>+-----+
|color|
+-----+
|    F|
|    E|
|    D|
|    J|
|    G|
|    I|
|    H|
+-----+<jupyter_text>Next, let us try to estimate an average price per carat. As you have noticed, the price column is an integer. This can result in a loss of precision as we do averaging. So first of all we will cast this column to double type:<jupyter_code>from pyspark.sql.types import DoubleType
from pyspark.sql.functions import *

# Convert Price column to type DoubleType
diamondsdf = diamonds.withColumn("price", diamonds["price"].cast(DoubleType()))<jupyter_output><empty_output><jupyter_text>We will use "groupby-aggregate function" to calculate the average. This creates a column with default name "avg(price)" which we rename to something easier to type. Finally, we order output by price in descending order:<jupyter_code># Calculate average price per carat value
carat_avgPrice = (diamondsdf
                  .groupBy("carat")
                  .avg("price")
                  .withColumnRenamed("avg(price)", "avgPrice")
                  .orderBy(desc("avgPrice")))

# View top10 highest average prices and corresponding carat value
carat_avgPrice.show(10)<jupyter_output>+-----+------------------+
|carat|          avgPrice|
+-----+------------------+
| 3.51|           18701.0|
| 2.67|           18686.0|
|  4.5|           18531.0|
| 5.01|           18018.0|
| 2.57|17841.666666666668|
|  2.6|           17535.0|
| 2.64|           17407.0|
| 4.13|           17329.0|
| 2.39|17182.428571428572|
| 2.71|           17146.0|
+-----+------------------+
only showing top 10 rows<jupyter_text>Analyzing CSV files in Python as RDDs (not recommended approach)In principle, one can use `RDD` to analyze structured data as well, but it seems to be less concenient, especially if the logic of your analysis can be expressed using SQL-like relational primitives.We will now convert our diamonds DataFrame into RDD:<jupyter_code># We can convert the DataFrame directly into an RDD
diamonds_rdd = diamonds.rdd
# View first 3 rows of the diamonds RDD
diamonds_rdd.take(3)<jupyter_output><empty_output><jupyter_text>You can now use RDD operations to analyze the data:<jupyter_code># Diamond counts by cuts
countByGroup = diamonds_rdd.map(lambda x: (x.cut, 1)).reduceByKey(lambda x,y: x+y)
print (countByGroup.collect())
# Distinct diamond clarities in dataset
distinctClarity = diamonds_rdd.map(lambda x: x.clarity).distinct()
print (distinctClarity.collect())
# Average price per diamond cut
avgPrice = diamonds_rdd.map(lambda x: (x.cut, float(x.price))).reduceByKey(lambda x,y: (x+y)/2)
print (avgPrice.collect())<jupyter_output>[('Ideal', 2756.7240663718817), ('Premium', 2756.654813661215), ('Good', 2755.647409027791), ('Very Good', 2756.7183661747795), ('Fair', 2743.567771968392)]<jupyter_text>Exercise: load a CSV file and analyze itUse what you have learned to load a set of `CSV` datasets. Open **load_csv_exercise.py** and follow the assignment therein.1. Actor1. Movie1. Actor playing in movie (relationships)and find movies where **<NAME>** played in.Save the answer in the `JSON` format.  Loading JSONThe best and probably the only reasonable way to load JSON files is using the Spark DataFrameReader.Spark SQL has built in support for reading in JSON files which contain a separate, self-contained JSON object per line. **Note: Multi-line JSON files are currently not compatible with Spark SQL.**<jupyter_code>testJsonData = spark.read.json("./data/json/test.json")
testJsonData.printSchema()
testJsonData.show()<jupyter_output>+---------+--------------------+---+-------+
|    array|                dict|int| string|
+---------+--------------------+---+-------+
|[1, 2, 3]|       [null,value1]|  1|string1|
|[2, 4, 6]|       [null,value2]|  2|string2|
|[3, 6, 9]|[extra_value3,val...|  3|string3|
+---------+--------------------+---+-------+<jupyter_text>Spark SQL can infer the schema automatically from your JSON data. To view the schema, use `printSchema`. Let's now try doing some basic queries to understand the dataset better.<jupyter_code># Count number of rows in dataset
print (testJsonData.count())<jupyter_output>3<jupyter_text>JSON data can contain nested data structures which can be accessed with a "."<jupyter_code>testJsonData.select('dict.key').show()<jupyter_output>+------+
|   key|
+------+
|value1|
|value2|
|value3|
+------+<jupyter_text>We can also perform DataFrame operations such as filtering queries according to some criteria:<jupyter_code>testJsonData.filter(testJsonData["int"] > 1).show()<jupyter_output>+---------+--------------------+---+-------+
|    array|                dict|int| string|
+---------+--------------------+---+-------+
|[2, 4, 6]|       [null,value2]|  2|string2|
|[3, 6, 9]|[extra_value3,val...|  3|string3|
+---------+--------------------+---+-------+<jupyter_text>Analyzing JSON files in Python with SQLAny DataFrame, including those created with JSON data, can be registered as a Spark SQL table to query with SQL.<jupyter_code># Create a Spark SQL temp table
# Note that temp tables are not global across clusters and will not persist across cluster restarts
testJsonData.registerTempTable("test_json")<jupyter_output><empty_output><jupyter_text>We can run any SQL queries on that table with Spark SQL:<jupyter_code>spark.sql("SELECT * FROM test_json").show()<jupyter_output>+---------+--------------------+---+-------+
|    array|                dict|int| string|
+---------+--------------------+---+-------+
|[1, 2, 3]|       [null,value1]|  1|string1|
|[2, 4, 6]|       [null,value2]|  2|string2|
|[3, 6, 9]|[extra_value3,val...|  3|string3|
+---------+--------------------+---+-------+
{'fraction_non_alphanumeric': 0.15268709191361124, 'fraction_numerical': 0.030554160388414534, 'mean_word_length': 3.738992463308211, 'pattern_counts': {'":': 0, '<': 52, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '15', 'dir': 'jupyter-structured-clean-dedup', 'id': '508498', 'license': 'Apache-2.0', 'n_tokens_mistral': 3971, 'n_tokens_neox': 3659, 'n_words': 1227, 'path': '2_LoadingData/LoadingData.ipynb', 'repo_name': 'felker/BigDataCourse', 'size': '11949'}