<jupyter_start><jupyter_text>Filter whole dataset using thresholds<jupyter_code>reviews = spark.read.csv(os.path.join(data_home, "ratings.csv"), header=True)
reviews.select(F.countDistinct("user_id")).show()
reviews.select(F.countDistinct("business_id")).show()
reviews.count()
user_thresh = 10
item_thresh = 1
dataset_name = "user_{}_item_{}".format(user_thresh, item_thresh)
active_users = (reviews.select("user_id", "business_id")
                .groupBy("user_id")
                .agg(F.count("*").alias("count_business"))
                .filter(col("count_business") >= user_thresh))
active_users.select(F.countDistinct("user_id")).show()
active_items = (reviews.select("user_id", "business_id")
                .groupBy("business_id")
                .agg(F.count("*").alias("count_user"))
                .filter(col("count_user") >= item_thresh))
active_items.select(F.countDistinct("business_id")).show()
filtered_reviews = (reviews
                    .join(active_users, on="user_id", how="inner")
                    .join(active_items, on="business_id", how="inner")
                    .drop("count_business", "count_user"))
filtered_reviews.show(5)
filtered_reviews.count()
windowSpec = Window \
    .partitionBy("user_id") \
    .orderBy(F.desc("date"))
test_reviews = filtered_reviews.withColumn("rank", F.dense_rank().over(windowSpec)).filter(col("rank") == 1).drop("rank")
train_reviews = filtered_reviews.withColumn("rank", F.dense_rank().over(windowSpec)).filter(col("rank") > 1).drop("rank")
train_file_dir, test_file_dir = [os.path.join(data_home, 
                                              "{}/yelp.{}".format(dataset_name, name))
                                for name in ["train", "test"]]
train_reviews.coalesce(1).write.csv(
    train_file_dir, 
    header=True, 
    mode="overwrite")
test_reviews.coalesce(1).write.csv(
    test_file_dir, 
    header=True, 
    mode="overwrite")
import subprocess
def collapse_dir_to_single_file(dir_, file_type="csv"):
    output = subprocess.check_output("ls {}/*.{}".format(dir_, file_type), shell=True).decode("utf-8").strip()
    file_name = dir_.split("/")[-1] + ".{}".format(file_type)
    file_name = os.path.abspath(os.path.join(dir_, "../", file_name))
    print("MOVING {} \nTO {}".format(output, file_name))
    os.system("cat {} > {}".format(output, file_name))
    os.system("rm -r {}".format(dir_))
    print("Done!\n")
for dir_ in [train_file_dir, test_file_dir]:
    collapse_dir_to_single_file(dir_)<jupyter_output>MOVING /media/ExtHDD01/recsys_data/yelp_dataset/tidy_data/user_10_item_1/yelp.train/part-00000-b5367686-6102-4cd1-89d4-09ceed6f3658-c000.csv 
TO /media/ExtHDD01/recsys_data/yelp_dataset/tidy_data/user_10_item_1/yelp.train.csv
Done!

MOVING /media/ExtHDD01/recsys_data/yelp_dataset/tidy_data/user_10_item_1/yelp.test/part-00000-6d5a8172-b730-45b1-9484-0d56cbb626d2-c000.csv 
TO /media/ExtHDD01/recsys_data/yelp_dataset/tidy_data/user_10_item_1/yelp.test.csv
Done!<jupyter_text>Generate business side info<jupyter_code>business_df = spark.read.json(os.path.join(data_home, "business.json")).filter(~F.isnull(col("categories")))

with open(os.path.join(data_home, "top_categories.json")) as f:
    top_categories = json.loads(f.read())
def find_top_category(str_cats):
    not_found = "NotFound"
    if not str_cats: return not_found
    
    cats = set(map(lambda x: x.strip(), str_cats.split(",")))
    intersections = cats.intersection(set(top_categories))
    
    return list(intersections)[0] if intersections else not_found
find_top_cat_udf = udf(find_top_category, StringType())

business_with_top = business_df.withColumn("top_category", find_top_cat_udf(col("categories")))
business_with_top.printSchema()
business_features = business_with_top.select("business_id", "top_category", "review_count", "stars", "state", "city")
business_side_info_dir = os.path.join(data_home, "side_info/business_features")
business_features.coalesce(1).write.csv(
    business_side_info_dir, 
    header=True,
    mode="overwrite")
collapse_dir_to_single_file(business_side_info_dir)
business_name_table = business_with_top.select("business_id", "name")
business_name_dir = os.path.join(data_home, "side_info/business_name_table")
business_name_table.coalesce(1).write.csv(
    business_name_dir,
    header=True,
    mode="overwrite")
collapse_dir_to_single_file(business_name_dir)<jupyter_output>MOVING /media/ExtHDD01/recsys_data/yelp_dataset/tidy_data/side_info/business_name_table/part-00000-b7674344-4372-4959-939d-f9e3b7656b9a-c000.csv 
TO /media/ExtHDD01/recsys_data/yelp_dataset/tidy_data/side_info/business_name_table.csv
Done!<jupyter_text>Generate user side information<jupyter_code>user_info = spark.read.json(os.path.join(data_home, "user.json"))
selected_cols = user_info.columns
selected_cols.remove("friends")
user_side_info = user_info.select(*selected_cols)
user_side_info.printSchema()
user_side_info = user_side_info.toPandas()
user_side_info_comp_related = user_side_info[[col for col in user_side_info.columns if "compliment_" in col]]

cnt_compliment = user_side_info_comp_related.sum(axis=1)

user_side_info = user_side_info[[col for col in user_side_info.columns if "compliment_" not in col]]

user_side_info["cnt_compliment"] = cnt_compliment

del user_side_info["name"]
def vec_time_delta(time_series, anchor_time):
    time_delta = np.array(anchor_time, dtype='datetime64[D]') - np.array(yelping_time, dtype='datetime64[D]') 
    return pd.Series(time_delta.astype(int))
yelping_time = pd.to_datetime(user_side_info.yelping_since)

user_side_info["yelp_days"] = vec_time_delta(yelping_time, "2019-12-31")

del user_side_info["yelping_since"]

user_side_info["elite_years"] = user_side_info.elite.fillna("").apply(lambda s: len(str(s).split(",")) if s else 0)

del user_side_info["elite"]
user_side_info.head()
user_side_info_path = os.path.join(data_home, "side_info/user.del_friends.csv")
user_side_info.to_csv(user_side_info_path, index=False)<jupyter_output><empty_output><jupyter_text>Sample negative test candidates<jupyter_code>candidate_num = 200
candidates_group = (business_with_top.groupBy("top_category", "state")
                    .agg(F.collect_list("business_id").alias("candidates")))
test_join_candidates = (business_with_top.select("business_id", "top_category", "state")
.join(test_reviews.select("business_id", "user_id"), on="business_id")
.join(candidates_group, on=["top_category", "state"])
.drop("top_category", "state"))
from pyspark.sql.functions import explode, rand
test_candidates = (test_join_candidates
                    .select( "user_id", "business_id", explode(test_join_candidates.candidates).alias("neg_cand"))
                    .filter(col("business_id") != col("neg_cand"))
                    .withColumn("random_num", rand()))
window = Window.partitionBy("user_id", "business_id").orderBy(F.desc("random_num"))
test_candidates = (test_candidates.withColumn("rank", F.dense_rank().over(window))
                   .filter(col("rank") <= candidate_num))
test_candidates.show()
test_ui_with_nested_candidates = (test_candidates.groupBy("user_id", "business_id")
                                                 .agg(F.collect_list("neg_cand").alias("candidates")))
test_ui_with_nested_candidates.show()
test_ui_with_candidates = test_candidates.select("user_id","business_id",col("neg_cand").alias("candidate_id"))
test_ui_with_candidates.printSchema()
train = pd.read_csv(os.path.join(data_home, dataset_name, "yelp.train.csv"))[["user_id", "business_id", "stars"]]
# remove samples within the train set
join_table = test_ui_with_candidates.merge(train, on=("user_id", "business_id"), how="left")
test_ui_with_candidates
filtered_candidates = join_table[pd.isna(join_table["stars"])][["user_id", "business_id", "candidate_id"]]
filtered_candidates
test_ground_truth = filtered_candidates[["user_id", "business_id"]]
test_ground_truth.drop_duplicates(inplace=True)
test_ground_truth = test_ground_truth.reset_index(drop=True)
test_ground_truth.rename(columns={"business_id":"candidate_id"}, inplace=True)
test_ground_truth
# combine test ground truth back
candidates = pd.concat([test_ground_truth, filtered_candidates[["user_id", "candidate_id"]]]).reset_index(drop=True)
candidates.to_parquet(os.path.join(data_home, dataset_name, "test_candidates.parquet"))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.146242221158449, 'fraction_numerical': 0.017831498324557205, 'mean_word_length': 5.761326860841424, 'pattern_counts': {'":': 1, '<': 16, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '595048', 'license': 'MIT', 'n_tokens_mistral': 3124, 'n_tokens_neox': 2940, 'n_words': 358, 'path': 'preprocessing/Generate testing set.ipynb', 'repo_name': 'HenryNebula/Personalization_Final_Project', 'size': '8356'}