I have a table of products and sub-products. It's quite a small table at 122 rows total and only 3 columns:
| backend | sub_product | product |
|---------|-------------|---------|
| conn_go | go_update | prod_go |
| conn_go | go_delete | prod_go |
| conn_go | go_edit | prod_go |
| conn_rv | rv_update | prod_rv |
| conn_mb | mb_update | prod_mb |
| conn_mb | mb_delete | prod_mb |
| conn_mb | mb_edit | prod_mb |
| conn_pr | pr_update | prod_pr |
| conn_pr | pr_edit | prod_pr |
| conn_ct | pol_edit | null |
....
I then have this daily partitioned usage table, which records how many times a user has used each sub-product. This table is a lot larger (186GB, 247MB per partition, 4.6M rows per partition):
| backend | yyyy_mm_dd | sub_product | x_id | user_id | count |
|---------|------------|-------------|------|---------|-------|
| conn_go | 2020-12-15 | go_update | 10 | 3422 | 1 |
| conn_go | 2020-12-15 | go_delete | 10 | 23445 | 2 |
| conn_go | 2020-12-15 | go_edit | 10 | 2243 | 2 |
| conn_rv | 2020-12-15 | rv_update | 10 | 245342 | 1 |
| conn_mb | 2020-12-15 | mb_update | 11 | 5464 | 3 |
| conn_mb | 2020-12-15 | mb_delete | 11 | 1424 | 2 |
| conn_mb | 2020-12-15 | mb_edit | 11 | 21454 | 2 |
| conn_pr | 2020-12-15 | pr_update | 12 | 224525 | 1 |
| conn_pr | 2020-12-15 | pr_edit | 12 | 22424 | 1 |
If a user_id has not used a sub_product on a day then there will be no row for that user_id / sub_product / yyyy_mm_dd combination in the usage table.
I would like to count, per x_id, the distinct count of user_ids who have used each product. Below is a the output I'm looking for, based on the above sample data.
| x_id | product | usage |
|------|---------|-------|
| 10 | prod_go | 3 |
| 10 | prod_rv | 1 |
| 10 | prod_mb | 0 |
| 10 | prop_pr | 0 |
| 10 | null | 0 |
| 11 | prod_go | 0 |
| 11 | prod_rv | 0 |
| 11 | prod_mb | 3 |
| 11 | prop_pr | 0 |
| 11 | null | 0 |
| 12 | prod_go | 0 |
| 12 | prod_rv | 0 |
| 12 | prod_mb | 0 |
| 12 | prop_pr | 1 |
| 12 | null | 1 |
What I've tried:
import pyspark.sql
from pyspark.sql import functions as sf
products = (
spark.table('my_schema.products')
.select('backend', 'sub_product', 'product')
)
usage = (
spark.table('my_schema.usage')
#.where(sf.col('yyyy_mm_dd')).between('2018-08-11', '2020-01-20')
.where(sf.col('count') > 0)
.where(sf.col('x_id').isNotNull())
.select('yyyy_mm_dd', 'x_id', 'user_id', 'backend', 'sub_product', 'count')
)
agg = (
products
.join(usage, on = ['backend', 'sub_product'], how = 'left')
.drop('count')
.groupby('x_id', 'product')
.agg(
sf.countDistinct('user_id')
)
)
agg.show(30,False)
The above Python runs for quite sometime (30+mins) and eventually times out:
org.apache.spark.SparkException: Job aborted due to stage failure: Task 157 in stage 136.0 failed 10 times, most recent failure: Lost task 157.9 in stage 136.0: ExecutorLostFailure (executor 3068 exited caused by one of the running tasks) Reason: Container marked as failed: Exit status: 143. Diagnostics: Container killed on request. Exit code is 143 Container exited with a non-zero exit code 143 Killed by external signal
Also, if I uncomment the where clause then I get another error:
"filter expression 'my_schema.usage.
yyyy_mm_dd' of type string is not a boolean
Q. Is there any way I can better optimize my code in order to get my desired output?
EDIT: Using a broadcast join greatly sped up the join and allows me to get an aggregated count. However, my code still doesn't produce the intended output from above:
agg = (
usage
.join(broadcast(products), on = ['backend', 'sub_product'], how = 'left')
.drop('count')
.where(sf.col('product').isNotNull())
.groupby('x_id', 'product')
.agg(
sf.countDistinct('user_id').alias('usage_ever')
)
)
agg.orderBy('x_id', 'product').show(500,False)
Problem: In my output, I do not get a list list of products per x_id i.e. I am missing rows when x_id has no users using a particular product (I would like to see 0 in such cases). Looking at the sample data, my output does not show a row for x_id = 10, product = prod_mb and usage = 0.