<jupyter_start><jupyter_text>Examine the schema<jupyter_code>df.printSchema()<jupyter_output>root
 |-- marketplace: string (nullable = true)
 |-- customer_id: integer (nullable = true)
 |-- review_id: string (nullable = true)
 |-- product_id: string (nullable = true)
 |-- product_parent: integer (nullable = true)
 |-- product_title: string (nullable = true)
 |-- product_category: string (nullable = true)
 |-- star_rating: integer (nullable = true)
 |-- helpful_votes: integer (nullable = true)
 |-- total_votes: integer (nullable = true)
 |-- vine: string (nullable = true)
 |-- verified_purchase: string (nullable = true)
 |-- review_headline: string (nullable = true)
 |-- review_body: string (nullable = true)
 |-- review_date: string (nullable = true)<jupyter_text>Rename columns<jupyter_code>df_review_id = df.select(['review_id', 'customer_id', 'product_id', 'product_parent', 'review_date'])
df_review_id.show(5)
df_review_id.printSchema()
from pyspark.sql.functions import to_date
review_df = df_review_id.withColumnRenamed('REVIEW_DATED', 'review_dated') 
products_df = df.select(['product_id', 'product_title'])
products_df = products_df.distinct()
total_rows = products_df.count()
print(f"The total nember of records after removing duplicates {total_rows}")
from pyspark.sql.functions import desc
customers_df = df.select(['customer_id'])\
  .groupby("customer_id")\
  .agg({"customer_id": "count"})\
  .orderBy(desc("count(customer_id)"))

customers_df.show()
total_rows = customers_df.count()
print(total_rows)
vine_df = df.select(['review_id', 'star_rating', 'helpful_votes', 'total_votes', 'vine'])
vine_df.show()
# Configuration for RDS instance
mode="append"
jdbc_url = "jdbc:postgresql://<insert endpoint>:5432/my_data_class_db"
config = {"user":"",
          "password": "",
          "driver":"org.postgresql.Driver"}
# Write DataFrame to table

review_df.write.jdbc(url=jdbc_url, table='review_id_table', mode=mode, properties=config)
products_df.write.jdbc(url=jdbc_url, table='products_table', mode=mode, properties=config)
customers_df.write.jdbc(url=jdbc_url, table='customers_table', mode=mode, properties=config)
vine_df.write.jdbc(url=jdbc_url, table='vine_table', mode=mode, properties=config)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1529255319148936, 'fraction_numerical': 0.002216312056737589, 'mean_word_length': 6.523333333333333, 'pattern_counts': {'":': 4, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '467170', 'license': 'ADSL', 'n_tokens_mistral': 772, 'n_tokens_neox': 763, 'n_words': 147, 'path': 'pt1_big_data.ipynb', 'repo_name': 'alexisbloor/big-data-challenge', 'size': '2256'}