<jupyter_start><jupyter_text>The steps below is optional. The code snippet installs pyspark module on the host and mounts and external storage location.<jupyter_code>!pip install pyspark

from google.colab import drive
drive.mount('/content/drive')<jupyter_output>Requirement already satisfied: pyspark in /usr/local/lib/python3.7/dist-packages (3.2.1)
Requirement already satisfied: py4j==0.10.9.3 in /usr/local/lib/python3.7/dist-packages (from pyspark) (0.10.9.3)
Drive already mounted at /content/drive; to attempt to forcibly remount, call drive.mount("/content/drive", force_remount=True).<jupyter_text>Importing necessary PySpark libraries<jupyter_code>from pyspark.sql import SparkSession
from pyspark.sql import types as T
import pyspark.sql.functions as F
from pyspark.sql.functions import *
from pyspark.sql.window import Window<jupyter_output><empty_output><jupyter_text>Initializing spark session<jupyter_code>spark = SparkSession.builder\
                    .appName('Adidas')\
                    .getOrCreate()
path = '/content/drive/MyDrive/Adidas/ol_cdump.json'

inp = spark.read.option('encoding','utf-8')\
                .json(path)
inp.printSchema()<jupyter_output>root
 |-- alternate_names: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- authors: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- author: struct (nullable = true)
 |    |    |    |-- key: string (nullable = true)
 |    |    |-- key: string (nullable = true)
 |    |    |-- type: string (nullable = true)
 |-- bio: string (nullable = true)
 |-- birth_date: string (nullable = true)
 |-- by_statement: string (nullable = true)
 |-- contributions: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- contributors: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- name: string (nullable = true)
 |    |    |-- role: string (nullable = true)
 |-- copyright_date: string (nullable = true)
 |-- covers: array (nullable = true)
 |    |-- element: long (containsNull = true)
 |-- created: struct (nullable = true)
 |    |-- type: string (nullable = true)
 |    |-- value: stri[...]<jupyter_text>The flatten method resolves complex data types such as ArrayType and StructType in order to un-nest the columns and 'flatten' the JSON file. This enables querying on the dataset for insight generation.<jupyter_code>def flatten(df):
    complex_fields = dict([
        (field.name, field.dataType) 
        for field in df.schema.fields 
        if isinstance(field.dataType, T.ArrayType) or isinstance(field.dataType, T.StructType)
    ])
    
    # print('complex_fields dict',complex_fields)

    qualify = list(complex_fields.keys())[0] + "_"

    while len(complex_fields) != 0:
        col_name = list(complex_fields.keys())[0]
        
        if isinstance(complex_fields[col_name], T.StructType):
            expanded = [F.col(col_name + '.' + k).alias(col_name + '_' + k) 
                        for k in [ n.name for n in  complex_fields[col_name]]
                       ]
            
            df = df.select("*", *expanded).drop(col_name)
    
        elif isinstance(complex_fields[col_name], T.ArrayType): 
            df = df.withColumn(col_name, F.explode_outer(col_name))
      
        complex_fields = dict([
            (field.name, field.dataType)
            for field in df.schema.fields
            if isinstance(field.dataType, T.ArrayType) or isinstance(field.dataType, T.StructType)
        ])
        
        
    for df_col_name in df.columns:
        df = df.withColumnRenamed(df_col_name, df_col_name.replace(qualify, ""))

    return df
flattened_df = flatten(inp)
flattened_df.printSchema()<jupyter_output>root
 |-- alternate_names: string (nullable = true)
 |-- bio: string (nullable = true)
 |-- birth_date: string (nullable = true)
 |-- by_statement: string (nullable = true)
 |-- contributions: string (nullable = true)
 |-- copyright_date: string (nullable = true)
 |-- covers: long (nullable = true)
 |-- death_date: string (nullable = true)
 |-- description: string (nullable = true)
 |-- dewey_decimal_class: string (nullable = true)
 |-- dewey_number: string (nullable = true)
 |-- download_url: string (nullable = true)
 |-- edition_name: string (nullable = true)
 |-- first_publish_date: string (nullable = true)
 |-- first_sentence: string (nullable = true)
 |-- full_title: string (nullable = true)
 |-- fuller_name: string (nullable = true)
 |-- genres: string (nullable = true)
 |-- ia_box_id: string (nullable = true)
 |-- ia_loaded_id: string (nullable = true)
 |-- isbn_10: string (nullable = true)
 |-- isbn_13: string (nullable = true)
 |-- isbn_invalid: string (nullable = true)
 |-- i[...]<jupyter_text>Cleaning the dataset by appying necessary filters - 1. Title should not be null2. The Number of pages should be greater than 203. The publish date should be later than 1950Fetching records which satisfy the above criteria<jupyter_code>filtered_inp = flattened_df.where('title IS NOT NULL')\
                           .where('number_of_pages > 20')\
                           .where('publish_date > 1950')<jupyter_output><empty_output><jupyter_text>Ques 1: Select all "Harry Potter" booksThe code below will fetch records where Harry Potter is properly spelled in the title column. However, we'll need to perform spell checks to make sure we are capturing records where 'Harry Potter' is incorrectly spelled. This can be acieved through Autocorrect Python package to look for strings which resemble 'Harry Potter'.<jupyter_code>filtered_inp.where('upper(title) like "%HARRY%POTTER%" OR upper(description) like "%HARRY%POTTER%"')\
            .show(truncate=False)<jupyter_output>+---------------+----+----------+-----------------------------------------------+-------------+--------------+------+----------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------------+------------+------------+------------+------------------+--------------+----------+-----------+-----------------+---------+------------+----------+-------+------------+---------------+------------------+---------------+------------------+----+--------+----+----------------------------------------------------------------------------------------------------------------------------------------+---------------+-----+-----------+------------+--------------------------------------------+----------+-------------+------+-------------------+---------------+---------------+------------+--------------+---------------------+------------+--------+------------[...]<jupyter_text>Ques 2: Get the book with the most pages<jupyter_code>max_pg = filtered_inp.agg(max(col('number_of_pages'))).head()[0]

filtered_inp.where('number_of_pages={0}'.format(max_pg))\
            .select('title','number_of_pages')\
            .distinct()\
            .show(truncate=False)

# filtered_inp.createOrReplaceTempView('filtered_inp')

# spark.sql('''
#     select distinct full_title, title, number_of_pages from filtered_inp where number_of_pages = (select max(number_of_pages) from filtered_inp)
# ''').show(truncate=False)<jupyter_output>+-----------------------------+---------------+
|title                        |number_of_pages|
+-----------------------------+---------------+
|<NAME>|48418          |
+-----------------------------+---------------+<jupyter_text>Ques 3: Find the Top 5 authors with most written books (assuming author in first position in the array, "key" field and each row is a different book)<jupyter_code>filtered_inp.withColumn('isbn_comb',coalesce('isbn_10','isbn_13'))\
            .where('authors_key IS NOT NULL')\
            .groupBy('authors_key')\
            .agg(count('isbn_comb').alias('cnt_isbn_comb'))\
            .orderBy('cnt_isbn_comb',ascending=False)\
            .limit(5)\
            .withColumn('authors_key',split(col('authors_key'),'/').getItem(2))\
            .show(truncate=False)<jupyter_output>+-----------+-------------+
|authors_key|cnt_isbn_comb|
+-----------+-------------+
|OL6551378A |51675        |
|OL6382015A |35400        |
|OL1056984A |10404        |
|OL6538270A |6432         |
|OL2695836A |5832         |
+-----------+-------------+<jupyter_text>Ques 4: Find the Top 5 genres with most books<jupyter_code>filtered_inp.withColumn('isbn_comb',coalesce('isbn_10','isbn_13'))\
            .where('genres IS NOT NULL')\
            .groupBy('genres')\
            .agg(count(col('isbn_comb')).alias('cnt_isbn_comb'))\
            .orderBy(col('cnt_isbn_comb'),ascending=False)\
            .limit(5)\
            .show(truncate=False)<jupyter_output>+--------------------+-------------+
|genres              |cnt_isbn_comb|
+--------------------+-------------+
|Essay               |53112        |
|Biography.          |26024        |
|Juvenile literature.|22491        |
|Statistics.         |17863        |
|Census, 1996.       |17700        |
+--------------------+-------------+<jupyter_text>Ques 5: Get the avg. number of pages<jupyter_code>filtered_inp.agg(round(avg('number_of_pages'),3).alias('avg_num_pages')).show(truncate=False)<jupyter_output>+-------------+
|avg_num_pages|
+-------------+
|280.142      |
+-------------+<jupyter_text>Ques 6: Per publish year, get the number of authors that published at least one book<jupyter_code>#confirming that publish date column has clean data

filtered_inp.where('length(publish_date) = 4').count()
filtered_inp.count()
filtered_inp.withColumn('isbn_comb',coalesce('isbn_10','isbn_13'))\
            .select(col('publish_date').cast('int'),'authors_key','isbn_comb')\
            .groupBy('publish_date','authors_key')\
            .agg(count('isbn_comb').alias('cnt_isbn_comb'))\
            .where('cnt_isbn_comb > 0')\
            .groupBy('publish_date')\
            .agg(count('authors_key').alias('cnt_authors'))\
            .show(truncate=False)<jupyter_output>+------------+-----------+
|publish_date|cnt_authors|
+------------+-----------+
|1959        |21         |
|1990        |1250       |
|1975        |293        |
|1977        |271        |
|2003        |1028       |
|2007        |1873       |
|1974        |267        |
|1955        |10         |
|2006        |920        |
|1978        |317        |
|1961        |22         |
|1952        |9          |
|1956        |15         |
|1997        |1813       |
|1988        |878        |
|1994        |1811       |
|1968        |161        |
|1951        |13         |
|1973        |273        |
|1979        |401        |
+------------+-----------+
only showing top 20 rows
{'fraction_non_alphanumeric': 0.24760309038443637, 'fraction_numerical': 0.02857674764963232, 'mean_word_length': 1.851380042462845, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '44684', 'license': 'MIT', 'n_tokens_mistral': 3314, 'n_tokens_neox': 3036, 'n_words': 688, 'path': 'Adidas_Case_Study_Pipeline.ipynb', 'repo_name': 'nb2195/open_lib_analysis_case_study', 'size': '10766'}