<jupyter_start><jupyter_text>Load the business users file to a dataframe<jupyter_code>df_bus = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").option("inferschema", "true").load("yelp_business.csv")
df_bus = df_bus.withColumnRenamed("name", "business_name").withColumnRenamed("business_id", "bus_business_id")
df_bus.printSchema()<jupyter_output>root
 |-- bus_business_id: string (nullable = true)
 |-- business_name: string (nullable = true)
 |-- neighborhood: string (nullable = true)
 |-- address: string (nullable = true)
 |-- city: string (nullable = true)
 |-- state: string (nullable = true)
 |-- postal_code: string (nullable = true)
 |-- latitude: double (nullable = true)
 |-- longitude: double (nullable = true)
 |-- stars: double (nullable = true)
 |-- review_count: integer (nullable = true)
 |-- is_open: integer (nullable = true)
 |-- categories: string (nullable = true)<jupyter_text>Convert the yelp review files by removing new lines in the comments<jupyter_code>%%bash

awk '/^"/ {if (f) print f; f=$0; next} {f=f FS $0} END {print f}' yelp_review.csv > yelp_review_fi.csv

%%bash
hdfs dfs -copyFromLocal /home/kkanagalananthapadm1/Scalable_Data/yelp_review_fi.csv<jupyter_output>copyFromLocal: `yelp_review_fi.csv': File exists<jupyter_text>Load the cleaned review file to a dataframe<jupyter_code>df_rev = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").option("inferschema", "true").load("yelp_review_fi.csv")
df_rev = df_rev.withColumnRenamed("stars", "review_stars").withColumnRenamed("useful", "review_useful").withColumnRenamed("date", "review_date").withColumnRenamed("cool", "review_cool").withColumnRenamed("funny", "review_funny")
df_rev.columns<jupyter_output><empty_output><jupyter_text>Load the users file to a dataframe<jupyter_code>df_user = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").option("inferschema", "true").load("yelp_user.csv")
df_user = df_user.withColumnRenamed("name", "user_name").withColumnRenamed("cool", "user_cool").withColumnRenamed("funny", "user_funny").withColumnRenamed("useful", "user_useful").withColumnRenamed("user_id", "user_user_id")
df_user.columns<jupyter_output><empty_output><jupyter_text>Check for null values in the 3 dataframes and remove null values if any<jupyter_code>from pyspark.sql.functions import isnan, when, count, col
#df_user.select([count(when(isnan(c), c)).alias(c) for c in df_user.columns]).show()
df_user.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in df_user.columns]).show()
from pyspark.sql.functions import isnan, when, count, col
#df_user.select([count(when(isnan(c), c)).alias(c) for c in df_user.columns]).show()
df_rev.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in df_rev.columns]).show()
df_rev = df_rev.na.drop()
df_rev.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in df_rev.columns]).show()
df_bus.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in df_bus.columns]).show()
df_bus = df_bus.na.drop()
df_bus.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in df_bus.columns]).show()<jupyter_output>+---------------+-------------+------------+-------+----+-----+-----------+--------+---------+-----+------------+-------+----------+
|bus_business_id|business_name|neighborhood|address|city|state|postal_code|latitude|longitude|stars|review_count|is_open|categories|
+---------------+-------------+------------+-------+----+-----+-----------+--------+---------+-----+------------+-------+----------+
|              0|            0|           0|      0|   0|    0|          0|       0|        0|    0|           0|      0|         0|
+---------------+-------------+------------+-------+----+-----+-----------+--------+---------+-----+------------+-------+----------+<jupyter_text>Join the 3 files to create meta table<jupyter_code>df_bus_rev = df_bus.join(df_rev,df_rev.business_id==df_bus.bus_business_id)
df_bus_rev_user = df_bus_rev.join(df_user,df_bus_rev.user_id==df_user.user_user_id)
df_bus_rev_user.printSchema()
df_bus_rev_user.columns<jupyter_output><empty_output><jupyter_text>Save the metatable as a csv with only one partition<jupyter_code>df_bus_rev_user.repartition(1).write.format("com.databricks.spark.csv").option("header", "true").save("final_merged_data.csv")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.23633879781420766, 'fraction_numerical': 0.004326047358834244, 'mean_word_length': 3.913870246085011, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '466808', 'license': 'MIT', 'n_tokens_mistral': 1543, 'n_tokens_neox': 1501, 'n_words': 251, 'path': 'Data_Merge.ipynb', 'repo_name': 'clintoncheang/Word-Distribution-Over-Yelp-Review', 'size': '4392'}