To do this effectively in Spark try the following technique:
1. Cache the SMALLER of the two datasets (which in this case is the 6,000 item product catalog). When you cache you will need to repartition the data such that there is data on every worker so you can use as many tasks as possible. Say you that your cluster has 25 workers with 4 cores per worker. You therefore need 4 x 25 = 100 partitions.
val products = sqlContext.table("products").repartition(100).cache()
2. Remember to call an action on the products DataFrame so that the caching occurs before the job starts.
products.take(1)
Check the SparkUI -> Storage tab to make sure that the products DF cache has been distributed on all nodes.
3. Disable broadcast joins temporarily. Broadcast joins don't work well for cartesian products because the workers get so much broadcast data they get stuck in an infinite garbage collection loop and never finish. Remember to turn this back on when the query finishes.
%sql set spark.sql.autoBroadcastJoinThreshold =0
4. Call join with the other table without using a join condition. You don't need to cache the larger table since it will take just as long to cache the bigger dataset as run the cross product.
val customers = table("customers")
val joined = customers.join(products)
5. Run an explain plan on the dataframe before executing to confirm you have a cartesian product operation.
joined.explain()
==PhysicalPlan== CartesianProduct :-ConvertToSafe :+-ScanParquetRelation[customer_key#45642,name#45643] InputPaths: dbfs:/tmp/customers +-ConvertToSafe +-ScanParquetRelation[product_key#45644,name#45645] InputPaths: dbfs:/tmp/products
6. Lastly save your result.
joined.write.save(...path...)
Happy Sparking!
