<jupyter_start><jupyter_text>0. Setup postgres connection<jupyter_code># postgresql connexion config
db_name="dbt_project"
host_name="127.0.0.1"
db_url=f"jdbc:postgresql://{host_name}:5432/{db_name}"
user="pliu"
password="<PASSWORD>"
driver="org.postgresql.Driver"
# note the driver value need to be changed if you use other database
# e.g. Mysql: com.mysql.jdbc.Driver
#     postgresql: org.postgresql.Driver
db_properties={"user": user, "password": password, "driver" : driver }<jupyter_output><empty_output><jupyter_text>1. Load customer csv to table csv<jupyter_code>customer_df = spark.read.options(header="True",inferSchema='True',delimiter=',').csv(customer)
customer_df.show()
customer_df.printSchema()
clean_customer_df=customer_df.withColumn("creation_date", to_timestamp("datetime_created", "yyyy-MM-dd HH:mm:ss"))\
           .withColumn("update_date",to_timestamp("datetime_updated","yyyy-MM-dd HH:mm:ss"))

clean_customer_df.show()
clean_customer_df.printSchema()
# write to postgres db server, with table name customers
clean_customer_df.write.jdbc(url=db_url,table="customers",mode='overwrite',properties=db_properties)<jupyter_output><empty_output><jupyter_text>You can check the generated customer table in postgres, the datetime_created column has type text (because it has string type in dataframe), the creation_date column has type timestamp (because it has timestamp type in dataframe)  2. Load order csv<jupyter_code>order_df=spark.read.options(header='True',inferSchema='True',delimiter=',').csv(orders)
order_df.show()
order_df.printSchema()
clean_order_df=order_df.withColumn("order_purchase_date",to_timestamp("order_purchase_timestamp","yyyy-MM-dd HH:mm:ss"))\
    .withColumn("order_approved_date",to_timestamp("order_approved_at","yyyy-MM-dd HH:mm:ss"))\
    .withColumn("order_deli_carrier_date",to_timestamp("order_delivered_carrier_date","yyyy-MM-dd HH:mm:ss")) \
    .withColumn("order_deli_customer_date",to_timestamp("order_delivered_customer_date","yyyy-MM-dd HH:mm:ss")) \
    .withColumn("order_estimated_deli_date",to_timestamp("order_estimated_delivery_date","yyyy-MM-dd HH:mm:ss")) \
    .drop("order_purchase_timestamp","order_approved_at","order_delivered_carrier_date","order_delivered_customer_date","order_estimated_delivery_date")

clean_order_df.show()
clean_order_df.printSchema()
clean_order_df.write.jdbc(url=db_url,table="orders",mode='overwrite',properties=db_properties)<jupyter_output><empty_output><jupyter_text>3. Load state csv<jupyter_code>state_df=spark.read.options(header='True',inferSchema='True',delimiter=',').csv(state)
state_df.show()
state_df.write.jdbc(url=db_url,table="states",mode='overwrite',properties=db_properties)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14107274063188832, 'fraction_numerical': 0.005143277002204262, 'mean_word_length': 10.25206611570248, 'pattern_counts': {'":': 2, '<': 18, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, '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': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '394973', 'license': 'Apache-2.0', 'n_tokens_mistral': 962, 'n_tokens_neox': 914, 'n_words': 124, 'path': 'Load_csv_to_postgres.ipynb', 'repo_name': 'pengfei99/DBTTutorial', 'size': '2720'}