Repository navigation
Expand file tree
/
Copy pathWorkers.scala
More file actions
75 lines (62 loc) · 2.66 KB
/
Copy pathWorkers.scala
File metadata and controls
75 lines (62 loc) · 2.66 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
package org.virtuslab.iskra.example.workers
import org.virtuslab.iskra.api.*
import functions.lit
@main def runExample(): Unit =
given spark: SparkSession = {
SparkSession
.builder()
.master("local")
.appName("workers")
.getOrCreate()
}
case class Worker(id: Long, firstName: String, lastName: String, yearsInCompany: Int)
case class Supervision(subordinateId: Long, supervisorId: Long)
case class Room(number: Int, name: String, desksCount: Int)
val workers = Seq(
Worker(3, "Bob", "Smith", 8),
Worker(13, "Alice", "Potter", 4),
Worker(38, "John", "Parker", 1),
Worker(21, "Julia", "Taylor", 3),
Worker(11, "Emma", "Brown", 6),
Worker(8, "Michael", "Johnson", 7),
Worker(18, "Natalie", "Evans", 4),
Worker(22, "Paul", "Wilson", 3),
Worker(44, "Daniel", "Jones", 1)
).toDF.asStruct
val supervisions = Seq(
44 -> 21, 22 -> 21, 38 -> 13, 11 -> 3, 21 -> 18, 13 -> 8, 3 -> 8, 18 -> 8
).map{ case (id1, id2) => Supervision(id1, id2) }.toDF.asStruct
workers.as("subordinates")
.leftJoin(supervisions).on($.subordinates.id === $.subordinateId)
.leftJoin(workers.as("supervisors")).on($.supervisorId === $.supervisors.id)
.select {
val salary = lit(4732) + $.subordinates.yearsInCompany * lit(214)
val supervisor = $.supervisors.firstName ++ lit(" ") ++ $.supervisors.lastName
($.subordinates.firstName, $.subordinates.lastName, supervisor, salary)
}
.where($.salary > lit(5000))
.show()
spark.stop()
import org.scalatest.funsuite.AnyFunSuite
import org.scalatest.matchers.should.Matchers.shouldEqual
import java.io.ByteArrayOutputStream
class ExampleTest extends AnyFunSuite:
test("Workers example") {
val outCapture = new ByteArrayOutputStream
Console.withOut(outCapture) { runExample() }
val result = new String(outCapture.toByteArray)
val expected = """|+---------+--------+---------------+------+
||firstName|lastName| supervisor|salary|
|+---------+--------+---------------+------+
|| Michael| Johnson| null| 6230|
|| Emma| Brown| Bob Smith| 6016|
|| Bob| Smith|Michael Johnson| 6444|
|| Alice| Potter|Michael Johnson| 5588|
|| Natalie| Evans|Michael Johnson| 5588|
|| Julia| Taylor| Natalie Evans| 5374|
|| Paul| Wilson| Julia Taylor| 5374|
|+---------+--------+---------------+------+
|
|""".stripMargin
result shouldEqual expected
}