How can I insert a PySpark dataframe into a database with a snowflake schema?

Question

With PySpark I'm computing a dataframe, how can I append this dataframe into my database, if this database has a snowflake schema?

How can I specify which way to split my dataframe in order to fit my CSV-like data into multiple joint tables?

My question is not specific to Pyspark, the same question could be asked about pandas.

Answer 1

To append a dataframe extracted from a CSV to a database consisting of a snowflake schema:

1. Extract the data from the snowflake schema.
2. Extract the new data from the external datasource.
3. Combine the two data sets.
4. Transform the combination to a set of dimension and fact tables to match the snowflake schema.
5. Load the transformed dataframes to the database, overwriting the existing data.

e.g. For a dataframe with the following schema, extracted from an external source:

StructType([StructField('customer_name', StringType()),
            StructField('campaign_name', StringType())])

def entrypoint(spark: SparkSession) -> None:
  extracted_customer_campaigns = extract_from_external_source(spark)

  existing_customers_dim, existing_campaigns_dim, existing_facts = (
    extract_from_snowflake(spark))

  combined_customer_campaigns = combine(existing_campaigns_dim,
                                        existing_customers_dim,
                                        existing_facts,
                                        extracted_customer_campaigns)

  new_campaigns_dim, new_customers_dim, new_facts = transform_to_snowflake(
    combined_customer_campaigns)

  load_snowflake(new_campaigns_dim, new_customers_dim, new_facts)


def combine(campaigns_dimension: DataFrame,
            customers_dimension: DataFrame,
            facts: DataFrame,
            extracted_customer_campaigns: DataFrame) -> DataFrame:
  existing_customer_campaigns = facts.join(
    customers_dimension,
    on=['customer_id']).join(
    campaigns_dimension, on=['campaign_id']).select('customer_name',
                                                    'campaign_name')

  combined_customer_campaigns = extracted_customer_campaigns.union(
    existing_customer_campaigns).distinct()

  return combined_customer_campaigns


def transform_to_snowflake(customer_campaigns: DataFrame) -> (
    DataFrame, DataFrame):
  customers_dim = customer_campaigns.select(
    'customer_name').distinct().withColumn(
    'customer_id', monotonically_increasing_id())

  campaigns_dim = customer_campaigns.select(
    'campaign_name').distinct().withColumn(
    'campaign_id', monotonically_increasing_id())

  facts = (
    customer_campaigns.join(customers_dim,
                            on=['customer_name']).join(
      campaigns_dim, on=[
        'campaign_name']).select('customer_id', 'campaign_id'))

  return campaigns_dim, customers_dim, facts

This is a simple functional approach. It maybe possible to optimise by writing deltas, rather than regenerating snowflake keys for each ETL batch.

In addition, if a separate external CSV were supplied containing records for deletion, this could be similarly extracted, then subtracted from the combined dataframe before transformation, in order to remove those existing records.

Finally, the question referred only to appending to a table. Additional steps would need to be manually added if merging/upserting were required as Spark itself does not support it.

Answer 2

You can do something like I describe on the code below. I am assuming that your csv has a similar structure as defined on df4. But I think you may not have the ids for customer_id, product_id and their groups. If that's the case, you can calculate them using that row_number windowing function (to have sequential numbers) or use the monotonically_increasing_id function as is shown to create df5

This solution is mostly based on PySpark and SQL, so if you are more familiar with traditional DW, you will understand better.

from pyspark.sql.functions import monotonically_increasing_id


#Creates input data. Only to rows to show how it should work
#The schema is defined on the single dataframe as 
# customer_id --> business key coming from transactional system
# customer_name --> just an attribute to show how it should behave
# customer_group_id --> an id that would match the group_id on the snowflake schema, as the idea is to group customers on groups (just as a sample)
# product_id --> another future dimension on the model having a snowflake schema
# product_group_id --> group id for products to group them on categories
df1 = spark.sql("""select 1 customer_id, 'test1' customer_name, 1 customer_group_id, 'group 1' customer_group_name, 
        1 product_id, 'product 1' product_name, 1 product_group_id, 'product group 1' product_group_name,
        987.5 sales
        """)

df2 = spark.sql("""select 2 customer_id, 'test2' customer_name, 1 customer_group_id, 'group 1' customer_group_name, 
        7 product_id, 'product 7' product_name, 1 product_group_id, 'product group 1' product_group_name,
        12345.5 sales
        """)

df3 = spark.sql("""select 2 customer_id, 'test2' customer_name, 1 customer_group_id, 'group 1' customer_group_name, 
        1 product_id, 'product 1' product_name, 1 product_group_id, 'product group 1' product_group_name,
        2387.3 sales
        """)

df4 = df1.union(df2).union(df3)

# Added an id on the df to be able to calculate the rest of the surrogate keys for dimensions
df5 = df4.withColumn("id",  monotonically_increasing_id())

# Registered dataframe to be able to query using SQL
df5.createOrReplaceTempView("df")

# Now create different dfs as the structure of the DW schema is
customer_group_df = spark.sql("""select customer_group_id, customer_group_name
            from df group by customer_group_id, customer_group_name""")

# I use the row_number because the monotonically increasing id function
# returns non sequential integers, but if you are good with that, it will be much faster
# Also another solution could be to use uuid as key (or other unique identifier providers)
# but that will depend on your requirements
customer_df = spark.sql("""select row_number() over (order by customer_id, customer_name, customer_group_id) as surkey_customer, customer_id customer_bk, 
            customer_name, customer_group_id
            from df group by customer_id, customer_name, customer_group_id """)

product_group_df =  spark.sql("""select product_group_id, product_group_name
            from df group by product_group_id, product_group_name""")

product_df =  spark.sql("""select row_number() over (order by product_id) as surkey_product, product_id product_bk, 
            product_name, product_group_id
            from df group by product_id, product_name, product_group_id""")

customer_df.show()
product_df.show()
df5.show()

# You can save those dfs directly on your model in the RBMS. Sorry as you are not defining the target DB I am not writing the code, 
# but should be done calling the save method of the dataframe pointing to Hive or to a JDBC where your DW model is
# You can find more info at https://stackoverflow.com/questions/30664008/how-to-save-dataframe-directly-to-hive or if 
# the target is a RDBMS https://stackoverflow.com/questions/46552161/write-dataframe-to-mysql-table-using-pyspark

# Now the tricky part is to calculate the surrogate keys of the fact table. The way to do it is to join back those df
# to the original dataframe. That can have performance issues, so please make sure that your data is 
# properly distributed (find the best approach to redistribute your dataframes on the nodes so that you reduce shuffling on the joins) 
# when you run 

customer_df.createOrReplaceTempView("customer_df")
product_df.createOrReplaceTempView("product_df")

fact_df = spark.sql("""
    select nvl(c.surkey_customer, -1) sk_customer, nvl(p.surkey_product, -1) sk_product, sales
    from
        df d left outer join customer_df c on d.customer_id = c.customer_bk   
            left outer join product_df p on d.product_id = p.product_bk
""").show()

# You can write the fact_df to your target fact table
# Be aware that to populate surrogate keys I am using nvl to assign the unknown member on the dimension. If you need
# that it also has to be present on the dimension table (customer and product, not group tables)

As you can see this solution uses simple snowflake schema. But the model can be more complex if you have Slowly Changing Dimensions Type 2 or other types of dimensional modeling

The output of that code is

+---------------+-----------+-------------+-----------------+
|surkey_customer|customer_bk|customer_name|customer_group_id|
+---------------+-----------+-------------+-----------------+
|              1|          1|        test1|                1|
|              2|          2|        test2|                1|
+---------------+-----------+-------------+-----------------+

+--------------+----------+------------+----------------+
|surkey_product|product_bk|product_name|product_group_id|
+--------------+----------+------------+----------------+
|             1|         1|   product 1|               1|
|             2|         7|   product 7|               1|
+--------------+----------+------------+----------------+

+-----------+-------------+-----------------+-------------------+----------+------------+----------------+------------------+-------+-----------+
|customer_id|customer_name|customer_group_id|customer_group_name|product_id|product_name|product_group_id|product_group_name|  sales|         id|
+-----------+-------------+-----------------+-------------------+----------+------------+----------------+------------------+-------+-----------+
|          1|        test1|                1|            group 1|         1|   product 1|               1|   product group 1|  987.5|          0|
|          2|        test2|                1|            group 1|         7|   product 7|               1|   product group 1|12345.5| 8589934592|
|          2|        test2|                1|            group 1|         1|   product 1|               1|   product group 1| 2387.3|17179869184|
+-----------+-------------+-----------------+-------------------+----------+------------+----------------+------------------+-------+-----------+

+-----------+----------+-------+
|sk_customer|sk_product|  sales|
+-----------+----------+-------+
|          1|         1|  987.5|
|          2|         2|12345.5|
|          2|         1| 2387.3|
+-----------+----------+-------+

Hope this helps