Spark Join operations

Author: sparkcodegeeks

src/main/scala/com/sparkbyexamples/spark/dataframe/join/CrossJoinExample.scala

@@ -0,0 +1,5 @@
+package com.sparkbyexamples.spark.dataframe.join
+
+class CrossJoinExample {
+
+}

src/main/scala/com/sparkbyexamples/spark/dataframe/join/InnerJoinExample.scala

@@ -0,0 +1,44 @@
+package com.sparkbyexamples.spark.dataframe.join
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.Inner
+
+object InnerJoinExample extends App {
+
+  val spark: SparkSession = SparkSession.builder()
+    .master("local[1]")
+    .appName("SparkByExamples.com")
+    .getOrCreate()
+
+  spark.sparkContext.setLogLevel("ERROR")
+
+  val emp = Seq((1,"Smith",-1,"2018","10","M",3000),
+    (2,"Rose",1,"2010","20","M",4000),
+    (3,"Williams",1,"2010","10","M",1000),
+    (4,"Jones",2,"2005","10","F",2000),
+    (5,"Brown",2,"2010","40","",-1),
+    (6,"Brown",2,"2010","50","",-1)
+  )
+  val empColumns = Seq("emp_id","name","superior_emp_id","year_joined","emp_dept_id","gender","salary")
+  import spark.sqlContext.implicits._
+  val empDF = emp.toDF(empColumns:_*)
+  empDF.show(false)
+
+  val dept = Seq(("Finance",10),
+    ("Marketing",20),
+    ("Sales",30),
+    ("IT",40)
+  )
+
+  val deptColumns = Seq("dept_name","dept_id")
+  val deptDF = dept.toDF(deptColumns:_*)
+  deptDF.show(false)
+
+
+  println("Inner join")
+  empDF.join(deptDF,empDF("emp_dept_id") ===  deptDF("dept_id"),"inner")
+    .show(false)
+
+  empDF.join(deptDF,empDF("emp_dept_id") ===  deptDF("dept_id"),Inner.sql)
+    .show(false)
+}

src/main/scala/com/sparkbyexamples/spark/dataframe/join/JoinMultipleColumns.scala

@@ -1,5 +1,60 @@
 package com.sparkbyexamples.spark.dataframe.join
 
-object JoinMultipleColumns {
+import org.apache.spark.sql.SparkSession
 
+object JoinMultipleColumns extends App {
+
+  val spark: SparkSession = SparkSession.builder()
+    .master("local[1]")
+    .appName("SparkByExamples.com")
+    .getOrCreate()
+
+  spark.sparkContext.setLogLevel("ERROR")
+
+  val emp = Seq((1,"Smith",-1,"2018",10,"M",3000),
+    (2,"Rose",1,"2010",20,"M",4000),
+    (3,"Williams",1,"2010",10,"M",1000),
+    (4,"Jones",2,"2005",10,"F",2000),
+    (5,"Brown",2,"2010",30,"",-1),
+    (6,"Brown",2,"2010",50,"",-1)
+  )
+  val empColumns = Seq("emp_id","name","superior_emp_id","branch_id","dept_id","gender","salary")
+  import spark.sqlContext.implicits._
+  val empDF = emp.toDF(empColumns:_*)
+  empDF.show(false)
+
+  val dept = Seq(("Finance",10,"2018"),
+    ("Marketing",20,"2010"),
+    ("Marketing",20,"2018"),
+    ("Sales",30,"2005"),
+    ("Sales",30,"2010"),
+    ("IT",50,"2010")
+  )
+
+  val deptColumns = Seq("dept_name","dept_id","branch_id")
+  val deptDF = dept.toDF(deptColumns:_*)
+  deptDF.show(false)
+
+  //Using multiple columns on join expression
+  empDF.join(deptDF, empDF("dept_id") === deptDF("dept_id") &&
+    empDF("branch_id") === deptDF("branch_id"),"inner")
+      .show(false)
+
+  //Using Join with multiple columns on where clause
+  empDF.join(deptDF).where(empDF("dept_id") === deptDF("dept_id") &&
+    empDF("branch_id") === deptDF("branch_id"))
+    .show(false)
+
+  //Using Join with multiple columns on filter clause
+  empDF.join(deptDF).filter(empDF("dept_id") === deptDF("dept_id") &&
+    empDF("branch_id") === deptDF("branch_id"))
+    .show(false)
+
+  //Using SQL & multiple columns on join expression
+  empDF.createOrReplaceTempView("EMP")
+  deptDF.createOrReplaceTempView("DEPT")
+
+  val resultDF = spark.sql("select e.* from EMP e, DEPT d " +
+    "where e.dept_id == d.dept_id and e.branch_id == d.branch_id")
+  resultDF.show(false)
 }

src/main/scala/com/sparkbyexamples/spark/dataframe/join/JoinMultipleDataFrames.scala

@@ -1,5 +1,35 @@
 package com.sparkbyexamples.spark.dataframe.join
 
-object JoinMultipleDataFrames {
+import org.apache.spark.sql.SparkSession
 
+object JoinMultipleDataFrames extends App {
+
+  val spark: SparkSession = SparkSession.builder()
+    .master("local[1]")
+    .appName("SparkByExamples.com")
+    .getOrCreate()
+
+  spark.sparkContext.setLogLevel("ERROR")
+
+  val emp = Seq((1,"Smith",-1,"2018","10","M",3000),
+    (2,"Rose",1,"2010","20","M",4000),
+    (3,"Williams",1,"2010","10","M",1000),
+    (4,"Jones",2,"2005","10","F",2000),
+    (5,"Brown",2,"2010","40","",-1),
+    (6,"Brown",2,"2010","50","",-1)
+  )
+  val empColumns = Seq("emp_id","name","superior_emp_id","year_joined","emp_dept_id","gender","salary")
+  import spark.sqlContext.implicits._
+  val empDF = emp.toDF(empColumns:_*)
+  empDF.show(false)
+
+  val dept = Seq(("Finance",10),
+    ("Marketing",20),
+    ("Sales",30),
+    ("IT",40)
+  )
+
+  val deptColumns = Seq("dept_name","dept_id")
+  val deptDF = dept.toDF(deptColumns:_*)
+  deptDF.show(false)
 }

src/main/scala/com/sparkbyexamples/spark/dataframe/join/SelfJoinExample.scala

@@ -1,5 +1,60 @@
 package com.sparkbyexamples.spark.dataframe.join
 
-object SelfJoinExample {
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.functions.col
+
+object SelfJoinExample extends App {
+
+  val spark: SparkSession = SparkSession.builder()
+    .master("local[1]")
+    .appName("SparkByExamples.com")
+    .getOrCreate()
+
+  spark.sparkContext.setLogLevel("ERROR")
+
+  val emp = Seq((1,"Smith",1,"10",3000),
+    (2,"Rose",1,"20",4000),
+    (3,"Williams",1,"10",1000),
+    (4,"Jones",2,"10",2000),
+    (5,"Brown",2,"40",-1),
+    (6,"Brown",2,"50",-1)
+  )
+  val empColumns = Seq("emp_id","name","superior_emp_id","emp_dept_id","salary")
+  import spark.sqlContext.implicits._
+  val empDF = emp.toDF(empColumns:_*)
+  empDF.show(false)
+
+  println("self join")
+  val selfDF = empDF.as("emp1").join(empDF.as("emp2"),
+    col("emp1.superior_emp_id") === col("emp2.emp_id"),"inner")
+  selfDF.show(false)
+
+  selfDF.select(col("emp1.emp_id"),col("emp1.name"),
+      col("emp2.emp_id").as("superior_emp_id"),
+      col("emp2.name").as("superior_emp_name"))
+    .show(false)
+
+  //Spark SQL self join with where clause
+  empDF.as("emp1").join(empDF.as("emp2")).where(
+    col("emp1.superior_emp_id") === col("emp2.emp_id"))
+  .select(col("emp1.emp_id"),col("emp1.name"),
+    col("emp2.emp_id").as("superior_emp_id"),
+    col("emp2.name").as("superior_emp_name"))
+    .show(false)
+
+  //Spark SQL self join with filter clause
+  empDF.as("emp1").join(empDF.as("emp2")).filter(
+    col("emp1.superior_emp_id") === col("emp2.emp_id"))
+    .select(col("emp1.emp_id"),col("emp1.name"),
+      col("emp2.emp_id").as("superior_emp_id"),
+      col("emp2.name").as("superior_emp_name"))
+    .show(false)
+
+
+  empDF.createOrReplaceTempView("EMP")
+  spark.sql("select emp1.emp_id,emp1.name," +
+    "emp2.emp_id as superior_emp_id, emp2.name as superior_emp_name " +
+    "from EMP emp1 INNER JOIN EMP emp2 on emp1.superior_emp_id == emp2.emp_id")
+    .show(false)
 
 }