How to Perform a Colocated Join
When two tables are partitioned on columns and colocated, it forces partitions having the same values for those columns in both tables to be located on the same SnappyData server. Colocating the data of two tables based on a partitioning column's value is a best practice if you frequently perform queries on those tables that join on that column. When colocated tables are joined on the partitioning columns, the join happens locally on the node where data is present, without the need of shuffling the data.
Code Example: ORDERS table is colocated with CUSTOMER table
A partitioned table can be colocated with another partitioned table by using the "COLOCATE_WITH" attribute in the table options.
For example, in the code snippet below, the ORDERS table is colocated with the CUSTOMER table. The complete source for this example can be found in the file ColocatedJoinExample.scala
Get a SnappySession:
val spark: SparkSession = SparkSession .builder .appName("ColocatedJoinExample") .master("local[*]") .getOrCreate val snSession = new SnappySession(spark.sparkContext)
Create Table Customer:
snSession.sql("CREATE TABLE CUSTOMER ( " + "C_CUSTKEY INTEGER NOT NULL," + "C_NAME VARCHAR(25) NOT NULL," + "C_ADDRESS VARCHAR(40) NOT NULL," + "C_NATIONKEY INTEGER NOT NULL," + "C_PHONE VARCHAR(15) NOT NULL," + "C_ACCTBAL DECIMAL(15,2) NOT NULL," + "C_MKTSEGMENT VARCHAR(10) NOT NULL," + "C_COMMENT VARCHAR(117) NOT NULL)" + "USING COLUMN OPTIONS (PARTITION_BY 'C_CUSTKEY')")
Create Table Orders:
snSession.sql("CREATE TABLE ORDERS ( " + "O_ORDERKEY INTEGER NOT NULL," + "O_CUSTKEY INTEGER NOT NULL," + "O_ORDERSTATUS CHAR(1) NOT NULL," + "O_TOTALPRICE DECIMAL(15,2) NOT NULL," + "O_ORDERDATE DATE NOT NULL," + "O_ORDERPRIORITY CHAR(15) NOT NULL," + "O_CLERK CHAR(15) NOT NULL," + "O_SHIPPRIORITY INTEGER NOT NULL," + "O_COMMENT VARCHAR(79) NOT NULL) " + "USING COLUMN OPTIONS (PARTITION_BY 'O_CUSTKEY', " + "COLOCATE_WITH 'CUSTOMER' )")
Perform a Colocate join:
// Selecting orders for all customers val result = snSession.sql("SELECT C_CUSTKEY, C_NAME, O_ORDERKEY, O_ORDERSTATUS, O_ORDERDATE, " + "O_TOTALPRICE FROM CUSTOMER, ORDERS WHERE C_CUSTKEY = O_CUSTKEY").collect()