<jupyter_start><jupyter_text>Introducing the DataFrames APIIn Spark, a DataFrame object consists of [Row](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.Row.html) objects and [Column](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.Column.html) objects. Concretely, each row of a Spark DataFrame  is an instance of the ```pyspark.sql.Row``` while each column is an instance of the ```pyspark.sql.Column``` class. We will look at  each of these classes in detail.  Creating DataFrames1. From Python objects2. External data sources3. Other Spark objects SchemasAlso, when creating DataFrames, you have the option to use a schema or not. A schema in Spark defines the column names and associated data types for a DataFrame. Most often, schemas come into play when you are reading structured data from an external data source. When a schema is not used, Spark has to infer the data type which can slow your application if you have a massive  dataset. Although schemas are more of DBMS language but they offer several advantages when dealing with large datasets:- Spark doesnt have to infer data types, so you get speed benefits.- Without a schema, Spark creates a separate job just to read a large portion of your file to ascertain the schema, which for a large data file can be expensive and time-consuming. As such, defining a schema will avoid this.- You can detect errors early if data doesn’t match the schema. Defining Schemas- Programmatically using Spark DataTypes - Using Data Definition Language (DDLs)  Spark DataFrame from Python objects<jupyter_code># Define schema using Spark DataTypes
schema = StructType([StructField("author_name", StringType(), False),
      StructField("book_title", StringType(), False),
      StructField("num_pages", IntegerType(), False)])

# Define Schema using DDL
schema = "author_name STRING, book_title STRING, num_pages INT"
# Define schema for our data using DDL
schema = "`Id` INT, `First` STRING, `Last` STRING, `Url` STRING,`Published` STRING, `Hits` INT, `Campaigns` ARRAY<STRING>"
# A simple statistic data
# in real life, we can get alot data in the o=form of Python objects and want to create SparkDataFrames
# for instance, data being downloaded from websites
data = [[1, "Jules", "Damji", "https://tinyurl.1", "1/4/2016", 4535, ["twitter","LinkedIn"]],
           [2, "Brooke","Wenig", "https://tinyurl.2", "5/5/2018", 8908, ["twitter",
    "LinkedIn"]],
           [3, "Denny", "Lee", "https://tinyurl.3", "6/7/2019", 7659, ["web",
    "twitter", "FB", "LinkedIn"]],
           [4, "Tathagata", "Das", "https://tinyurl.4", "5/12/2018", 10568,
    ["twitter", "FB"]],
           [5, "Matei","Zaharia", "https://tinyurl.5", "5/14/2014", 40578, ["web",
    "twitter", "FB", "LinkedIn"]],
           [6, "Reynold", "Xin", "https://tinyurl.6", "3/2/2015", 25568,
    ["twitter", "LinkedIn"]]
          ]

# Create a SparkSession
# spark = SparkSession.builder.master("local[*]").appName("DataFrameFromPythonObj").getOrCreate()
spark=SparkSession.builder.appName("intro").master("local[*]").config("spark.driver.bindAddress", "127.0.0.1").getOrCreate()

# Create a DataFrame using the schema defined above
sdf = spark.createDataFrame(data, schema)
# Show the DataFrame; it should reflect our table above blogs_df.show()
# Print the schema used by Spark to process the DataFrame
print(sdf.printSchema())<jupyter_output>root
 |-- Id: integer (nullable = true)
 |-- First: string (nullable = true)
 |-- Last: string (nullable = true)
 |-- Url: string (nullable = true)
 |-- Published: string (nullable = true)
 |-- Hits: integer (nullable = true)
 |-- Campaigns: array (nullable = true)
 |    |-- element: string (containsNull = true)

None<jupyter_text>EXERCISE-1: READ CSV WITH SCHEMA1. Use Spark documentation on how to read from file with a define schema. Note, the schema is what we arleady defined above. The data above has been saved as ```blog_simple_dataset.csv```. Read it as a Spark DataFrame with schema. Answer this question in the next cell.2. Define schema for the ```activity_raw_data.csv``` use string for the datetime column3. Load the dataset with and without schema using the functions defined below. Compare the loading times. Answer this question by completing the functions defined below and calling them.  Spark DataFrame from external data sourcesThe most common way (which we have already seen) is to load data from exteernal data sources and Spark supports numerous data stores. Spark reads data  through the ```DataFrameReaderobject```. Please look at the documeentation [here](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.DataFrameReader.csv.html) to see all data sources that the Spark  ```DataFrameReaderobject``` supports.<jupyter_code>from IPython.display import Image
Image("SparkConnectors.png")
def timefn(fn):
    """
    Function for recording running time of a function
    """
    @wraps(fn)
    def measure_time(*args, **kwargs):
        t1 = time.time()
        result = fn(*args, **kwargs)
        t2 = time.time()
        print("@timefn:" + fn.__name__ + " took " + str(t2 - t1) + " seconds")
        return result
    return measure_time
@timefn
def load_with_schema(large_csv):
    # define the schema here
    # you can load part of the file with pandas (just a few rows)
    # to remind yourself of the data types
    schema = "`ACTIVITY_ID` INT"
    spark = SparkSession.builder.master("local[*]").appName("ReadWithChema").getOrCreate()
    # Now read the data 
    sdf =  spark.read.schema(schema).csv(large_csv)
large_csv = "/home/user/Desktop/big data analytic/activity_log_raw.csv"
pd.read_csv(large_csv,nrows=100)
load_with_schema(large_csv)
@timefn
def load_without_schema(large_csv):
    spark = SparkSession.builder.master("local[*]").appName("DataFrameFromPythonObj").getOrCreate()
    sdf = spark.read.csv(large_csv, header=True)
load_with_schema("/home/user/Desktop/big data analytic/activity_log_raw.csv")
load_without_schema("/home/user/Desktop/big data analytic/activity_log_raw.csv")<jupyter_output>@timefn:load_without_schema took 0.10169053077697754 seconds<jupyter_text>Columns and Expressions in  DataFramesIn Spark DataFrames, columns behave like pandas DataFrames in several ways but they also behave different. You can list all the columns by their names, and you can perform operations on their values using relational or computational expressions. - [Column](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.Column.html) is the name of the object, which has many import methods such as describe  while [col()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.functions.col.html) is a standard built-in function that returns a Column.We need to use the col() and expr() function available in pyspark,sql.functions() for many operations such as:- Add, rename columns- Subset data based on columns- Access columns to compute stats on them-  Access columns to compute operations on them such as sorting  Add a new column using expr and colIn order to add a new column in a Spark DataFrame, we use the ```DataFrame.withColumn(new_col_name, expression_to_compute_new_col)```<jupyter_code>from pyspark.sql.functions import *
csv_fpath = "/home/user/Desktop/big data analytic/activity_log_raw.csv"
spark = SparkSession.builder.master("local[*]").appName("DataFrameFromPythonObj").getOrCreate()
sdf = spark.read.csv(csv_fpath, header=True)
# use expr
sdf2 = sdf.withColumn("new_col", (expr("ACTIVITY_ID > 10000")))
sdf2.show()
l = ['S','N']
data["Gender"].isin(["Male"])
sdf2.filter('STATUS', is
sdf2.filter('STATUS'.isin())
# use the col function which I prefer over the expr col("Hits")
sdf2 = sdf.withColumn("new_col", col("ACTIVITY_ID") > 10000)
sdf2.show()
df1 = sdf2.drop('SID','STATUS')
df1.show()<jupyter_output>+-----------+--------------------+-------+
|ACTIVITY_ID|       ACTIVITY_TIME|new_col|
+-----------+--------------------+-------+
|       1291|13-APR-15 10.33.4...|  false|
|       1286|13-APR-15 10.33.4...|  false|
|       1285|13-APR-15 10.33.4...|  false|
|       1284|13-APR-15 10.33.4...|  false|
|       1288|13-APR-15 10.33.4...|  false|
|       1293|13-APR-15 10.33.4...|  false|
|         10|13-APR-15 10.33.3...|  false|
|         10|13-APR-15 10.33.3...|  false|
|         10|13-APR-15 10.33.3...|  false|
|       1269|13-APR-15 10.33.2...|  false|
|       1268|13-APR-15 10.33.2...|  false|
|       1267|13-APR-15 10.33.2...|  false|
|       1266|13-APR-15 10.33.2...|  false|
|       1265|13-APR-15 10.33.2...|  false|
|       1264|13-APR-15 10.33.2...|  false|
|       1263|13-APR-15 10.33.2...|  false|
|       1262|13-APR-15 10.33.2...|  false|
|       1261|13-APR-15 10.33.2...|  false|
|       1260|13-APR-15 10.33.2...|  false|
|       1259|13-APR-15 10.33.2...|  false|
+----------[...]<jupyter_text>Subset data  based on a few columnsIn order to access a single or multiple columns, we use the ```select()``` function on the DataFrame<jupyter_code>sdf3 = sdf.select('ACTIVITY_TIME', 'STATUS')
sdf3
sdf3.withColumn('new_column', lit(None).cast(StringType())).show()<jupyter_output>+--------------------+------+----------+
|       ACTIVITY_TIME|STATUS|new_column|
+--------------------+------+----------+
|13-APR-15 10.33.4...|     S|      null|
|13-APR-15 10.33.4...|     S|      null|
|13-APR-15 10.33.4...|     S|      null|
|13-APR-15 10.33.4...|     S|      null|
|13-APR-15 10.33.4...|     S|      null|
|13-APR-15 10.33.4...|     S|      null|
|13-APR-15 10.33.3...|     N|      null|
|13-APR-15 10.33.3...|     R|      null|
|13-APR-15 10.33.3...|     N|      null|
|13-APR-15 10.33.2...|     S|      null|
|13-APR-15 10.33.2...|     S|      null|
|13-APR-15 10.33.2...|     S|      null|
|13-APR-15 10.33.2...|     S|      null|
|13-APR-15 10.33.2...|     S|      null|
|13-APR-15 10.33.2...|     S|      null|
|13-APR-15 10.33.2...|     S|      null|
|13-APR-15 10.33.2...|     S|      null|
|13-APR-15 10.33.2...|     S|      null|
|13-APR-15 10.33.2...|     S|      null|
|13-APR-15 10.33.2...|     S|      null|
+--------------------+------+----------+
only showing top[...]<jupyter_text>**EXERCISE-2:**1. Check  if these statements: df.select(expr("ACTIVITY_TIME")).show(2), df.select(col("ACTIVITY_TIME")).show(2)and df.select("ACTIVITY_TIME").show(2) will provide  the same output. Replace df with name of your Spark DataFrame.2. Create a new DataFrame using expr to get only those rows where STATUS is "S"Note that expr() just perfoms the operation, it doesnt filter our the rows which evaluate to false.2. Sort DataFrame: use the col function to sort the DataFrame on "SID" column<jupyter_code># YOUR CODE
sdf.sort(col("SID")).show()<jupyter_output><empty_output><jupyter_text>RowsA row in Spark is a generic Row object, containing one or more columns. Each column may be of the same data type (e.g., integer or string), or they can have different types (integer, string, map, array, etc.). Because Row is an object in Spark and an ordered collection of fields, you can instantiate a Row the same way we instantiate any object. Consequently, you can collect Row objects in a list and create a Spark DataFrame.<jupyter_code>from pyspark import Row
row = Row(name="Alice", age=11)
rows = [Row(name="<NAME>", state="CA"), Row(name="<NAME>", state="CA")]
spark_df_from_rows = spark.createDataFrame(rows)
spark_df_from_rows.show()<jupyter_output><empty_output><jupyter_text>**EXERCISE-3:** Creating a Spark DataFrame with Rows. Please complete the function below and call it.<jupyter_code>def convert_json_to_spark_with_rows(json_file):
    # create  a list to hold all Row objects
    rows = YOUR CODE
    for idx, row in df.iterrows():
        # get lon and lat from the coord column using indexing, dict key access
        x = row['coord']['lon']       
        y = row['coord']['lat']
        # create the Row object here 
        srow = YOUR CODE
        
        # append this row object to the list
        YOUR CODE
    
    # When creating Spark DataFrame this way, its better to use schema to avoid troubles
    # create a schema for this data here, use DOUBLE as data type for lon and lat
    schema = YOUR CODE
    
    # use spark.createDataFrame() here
    # if yiu get errors, use the option verifySchema=False
    spark_df = YOUR CODE
    
    # use show() statement to show the DataFrame
    # use show() with print to ensure we see the outputs
    YOUR CODE
jsonfile = "../data/city.list.json"
convert_json_to_spark_with_rows(jsonfile)
def add_date_and_filter(csv_file, date_format, ref_date):
    """
    Create a dataframe, add date and filter out events based on ref date
    :return:
    """
    df = pd.read_csv(csv_file)
    str_time_col = 'cdr datetime'

    # convert date string to Python datetime
    # please extract only the Year, month and day from the date string using
    # indexing, please use solutions from first assignment to achieve this
    #f = lambda x: pd.to_datetime(x[:10], format=date_format)
    df['cdr_date'] = df[str_time_col].apply(lambda x: pd.to_datetime(str(x)[:9],format=date_format))     ########YOUR CODE

    # please retrieve all events older than or equal to the reference date
    # please use the query function for pandas DataFrame like below
    df.query('cdr_date <= @ref_date', inplace=True)

    if df.shape[0]:
        return df
    else:
        return 'NA'
def get_events_before_jul13(csv_folder, ref_date, date_format, num_csv_files):
    """
    In this function, we will use the map function to run the
    add_date_and_filter above on a list of CSV_files
    :param csv_folder:
    :return:
    """
    list_csv = [f for f in csv_folder.iterdir() if f.suffix == '.csv']
    # use the function random.choices() to select only a sample of the CSVs  for fast processing
    # set k to num_csv_files in the function
    list_csv =  random.choices(list_csv, k=num_csv_files)
    time_format = "%Y%m%d"

    # Use  the datetime.strptime() to convert the string ref_date to Python datetime object
    # This function requires two args, the other one is date format
    ref_date = datetime.strptime(ref_date,  date_format)

    # Run map with multiple iterables(r.g., two lists)
    # Note that input-1 is the list of CSV we generated above
    # Prepare input-2 and input-3 as below. Please use the  multiplication
    # operator on a single list item to repeat the same element multiple
    # times in a list as we did in the first assignment
    list_date_format = [date_format]*len(list_csv)
    list_date_ref_date = [ref_date]*len(list_csv)

    # now run the map function with the three input lists we have defined
    results = map(add_date_and_filter,list_csv,list_date_format,list_date_ref_date)

    # write code to identify elements in the list which arent s 'NA'
    # You can use a list comprehension with if condition. To identify elements
    # which are 'NA', you can use isinstance() function to detect data type
    no_na_results = []
    print('I found {} non NA results'.format(len(no_na_results)))

    return no_na_results
from pathlib import Path as p
import random
from datetime import datetime
link =  p("/home/user/Desktop/big data analytic/test")
get_events_before_jul13(link,'20180713','%Y%m%d',2)<jupyter_output>I found 0 non NA results<jupyter_text>Common DataFrames Operations<jupyter_code># TO BE CONTINUED
csv_fpath = "/Users/dmatekenya/Desktop/TMP/data/activity_log_raw.csv"
spark = SparkSession.builder.master("local[*]").appName("DataFrameFromPythonObj").getOrCreate()
sdf = spark.read.csv(csv_fpath, header=True)
sdf2 = sdf.select('ACTIVITY_TIME', 'STATUS')
sdf2.show()
sdf3 = sdf.select('ACTIVITY_TIME', 'STATUS')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13578599567924768, 'fraction_numerical': 0.03844198754606684, 'mean_word_length': 3.464964539007092, 'pattern_counts': {'":': 0, '<': 32, '<?xml version=': 0, '>': 33, 'https://': 11, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '392767', 'license': 'MIT', 'n_tokens_mistral': 5468, 'n_tokens_neox': 5021, 'n_words': 1636, 'path': 'Spark-Structured-APIs.ipynb', 'repo_name': 'Louis-Mozart/Louis-Mozart.github.io', 'size': '15750'}