In [32]:
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
In [33]:
reviews_parquet_path = r"..\Yelp JSON\yelp_reviews.parquet"
businesses_parquet_path = r"..\Yelp JSON\yelp_businesses.parquet"
users_parquet_path = r"..\Yelp JSON\yelp_users.parquet"
output_path = r"..Yelp JSON\yelp_merged_PA_NJ.parquet"
states_included = ['PA', 'NJ']
In [34]:
# Loading smaller "lookup" tables
print("loading lookup tables...")
df_businesses = pd.read_parquet(businesses_parquet_path)
df_businesses = df_businesses[df_businesses['state'].isin(states_included)]
state_selected_business_id = list(df_businesses['business_id'].unique())
df_users = pd.read_parquet(users_parquet_path)
print("lookup tables loaded...")
loading lookup tables... lookup tables loaded...
In [35]:
# To prevent pandas from creating ambiguous columns like 'name_x', 'name_y'
# During merges, we explicitly rename columns that exist in multiple tables.
print("Renaming conflicting columns in lookup tables...")
df_businesses = df_businesses.rename(columns={
'name': 'business_name',
'stars': 'business_avg_stars',
'review_count': 'business_review_count'
})
df_users = df_users.rename(columns={
'name': 'user_name',
'review_count': 'user_review_count',
'average_stars': 'user_avg_stars',
'useful': 'user_total_useful_votes',
'funny': 'user_total_funny_votes',
'cool': 'user_total_cool_votes'
})
print("Column renaming complete...")
Renaming conflicting columns in lookup tables... Column renaming complete...
In [36]:
# Joining reviews table with businesses and users tables
print("Loading reviews and merging with businesses and users...")
df_reviews = pd.read_parquet(reviews_parquet_path)
df_reviews = df_reviews[df_reviews['business_id'].isin(state_selected_business_id)]
merged_chunk = df_reviews.merge(df_businesses, on='business_id', how='left') \
.merge(df_users, on='user_id', how='left')
# The output is disabled to avoid writing to disk. Uncomment the next line to save the merged DataFrame.
# merged_chunk.to_parquet(output_path)
print(f"Successfully created merged file with selected states at: {output_path}")
Loading reviews and merging with businesses and users... Successfully created merged file with selected states at: ..Yelp JSON\yelp_merged_PA_NJ.parquet
In [37]:
print("Unclean Categories:\n\n", merged_chunk['categories'].head(), "\n\n")
# The merged chunk df has unclean data in the "categorys" column.
# Here we clean and parse through each row and create a value count of each category to find the top most common. We will keep the most common categories.
merged_chunk['categories'] = (
merged_chunk['categories']
.astype(str)
.str.replace(r'\s*,\s*', ',', regex=True) # remove spaces around commas
.str.replace(r'\s+', ' ', regex=True) # collapse extra spaces
.str.strip() # trim leading/trailing spaces
.str.lower() # normalize case
)
print("Clean Categories:\n\n", merged_chunk['categories'].head(), "\n\n")
Unclean Categories: 0 Restaurants, Breakfast & Brunch, Food, Juice B... 1 Active Life, Cycling Classes, Trainers, Gyms, ... 2 Halal, Pakistani, Restaurants, Indian 3 Mediterranean, Restaurants, Seafood, Greek 4 Beer Bar, Bars, American (New), Gastropubs, Re... Name: categories, dtype: object Clean Categories: 0 restaurants,breakfast & brunch,food,juice bars... 1 active life,cycling classes,trainers,gyms,fitn... 2 halal,pakistani,restaurants,indian 3 mediterranean,restaurants,seafood,greek 4 beer bar,bars,american (new),gastropubs,restau... Name: categories, dtype: object
In [38]:
# Here we filter the merged_chunk DataFrame to keep only the rows that match our selected categories.
print("Filtering by categories...")
categories_to_keep = ['clothing', 'health & medical', 'auto repair', 'arts & entertainment']
df_filtered_by_categories = []
for i in categories_to_keep:
cat_df = merged_chunk[merged_chunk['categories'].str.contains(i, case=False, na=False)]
df_filtered_by_categories.append(cat_df)
df_cleaned_categories = pd.concat(df_filtered_by_categories)
# Standardize category names and group less common categories into 'other'. Then we filter the DataFrame to keep only the selected categories.
for cat in categories_to_keep:
mask = merged_chunk['categories'].str.contains(cat, case=False, na=False)
merged_chunk.loc[mask, 'categories'] = cat
merged_chunk.loc[~merged_chunk['categories'].isin(categories_to_keep), 'categories'] = 'other'
filtered_merged_chunk = merged_chunk[merged_chunk['categories'].isin(categories_to_keep)]
print(f"Finished filtering categoryies...")
print("total reviews:", df_cleaned_categories.shape[0],"\n")
print(filtered_merged_chunk['categories'].value_counts())
Filtering by categories... Finished filtering categoryies... total reviews: 174646 categories arts & entertainment 69164 health & medical 55257 auto repair 34949 clothing 13904 Name: count, dtype: int64
In [39]:
# Define the bounding box for the Greater Philadelphia area
print("Applying bounding box filter for Greater Philadelphia area...")
bbox = (
filtered_merged_chunk['latitude'].between(39.85, 40.03, inclusive='both') &
filtered_merged_chunk['longitude'].between(-75.30, -75.07, inclusive='both')
)
final_df = filtered_merged_chunk.loc[bbox].copy()
print("Bounding box filter applied...")
print("Total Reviews after bounding box filter:", final_df.shape[0])
Applying bounding box filter for Greater Philadelphia area... Bounding box filter applied... Total Reviews after bounding box filter: 80280
In [40]:
# The output is disabled to avoid writing to disk. Uncomment the next lines to save the parquet files.
output_path_final = r"..\Yelp JSON\yelp_reviews_filtered.parquet"
print("Saving the filtered merged chunk to Parquet file...")
# final_df.to_parquet(output_path_final, index=False)
print("Filtered merged chunk saved successfully...")
Saving the filtered merged chunk to Parquet file... Filtered merged chunk saved successfully...