Scala Project: Data Pipeline
A data pipeline is a chain of steps that turns raw records into a report. This lesson builds one over records held in code.
Try holds a parse failure as a value
A good row gives its amount. A bad row gives the fallback instead of throwing.
case class Sale(item: String, amount: Int)
def parse(text: String): scala.util.Try[Sale] =
val parts = text.split(",")
scala.util.Try(Sale(parts(0), parts(1).toInt))
@main def main(): Unit =
println(parse("apples,30").map(s => s.amount).getOrElse(0))
println(parse("pears,xx").map(s => s.amount).getOrElse(0))
Validation collects every problem at once
One bad record can carry two problems. Validate returns both in one list.
case class Sale(item: String, amount: Int)
def validate(s: Sale): List[String] =
List(
if s.item.isEmpty then Some("empty item") else None,
if s.amount <= 0 then Some("bad amount") else None
).flatten
@main def main(): Unit =
println(validate(Sale("", -5)))
println(validate(Sale("apples", 30)))
A view filters, maps, and takes two names
The view chains filter and map without building lists between steps. Forcing it with toList gives the first two names.
case class Sale(item: String, amount: Int)
@main def main(): Unit =
val sales = List(Sale("apples", 30), Sale("pears", 20), Sale("plums", 50), Sale("peaches", 40))
val names = sales.view.filter(s => s.amount >= 25).map(s => s.item).take(2).toList
println(names)
Grouping totals each category separately
GroupBy collects sales by category. Sorted keys give a fixed print order, and each lookup totals one group.
case class Sale(category: String, item: String, amount: Int)
@main def main(): Unit =
val sales = List(Sale("fruit", "apples", 30), Sale("fruit", "plums", 50), Sale("veg", "carrots", 20), Sale("veg", "potatoes", 40))
val groups = sales.groupBy(s => s.category)
println(groups.keys.toList.sorted.mkString(", "))
println(groups("fruit").map(s => s.amount).sum)
println(groups("veg").map(s => s.amount).sum)
A given Ordering sorts sales by amount
The given Ordering compares amounts. Sorted then lists the smallest sale first.
case class Sale(item: String, amount: Int)
given Ordering[Sale] with
def compare(a: Sale, b: Sale): Int = a.amount.compare(b.amount)
@main def main(): Unit =
val sales = List(Sale("apples", 30), Sale("pears", 20), Sale("plums", 50))
println(sales.sorted.map(s => s.item).mkString(", "))
Example
case class Sale(category: String, item: String, amount: Int)
given Ordering[Sale] with
def compare(a: Sale, b: Sale): Int = b.amount.compare(a.amount)
def parse(text: String): Either[List[String], Sale] =
val parts = text.split(",")
if parts.length != 3 then Left(List("bad row: " + text))
else
val category = parts(0)
val item = parts(1)
val amountOpt = scala.util.Try(parts(2).toInt).toOption
val errors = List(
if item.isEmpty then Some("empty item") else None,
if amountOpt.isEmpty then Some("bad amount: " + parts(2)) else if amountOpt.get <= 0 then Some("bad amount: " + parts(2)) else None
).flatten
if errors.isEmpty then Right(Sale(category, item, amountOpt.get)) else Left(errors)
@main def main(): Unit =
val raw = List("fruit,apples,30", "fruit,plums,50", "veg,carrots,20", "veg,potatoes,40", "fruit,pears,-5", "veg,,25")
val parsed = raw.map(parse)
val errors = parsed.collect { case Left(errs) => errs }.flatten
val good = parsed.collect { case Right(s) => s }
val kept = for s <- good if s.amount >= 25 yield s
val groups = kept.groupBy(s => s.category)
println("Errors: " + errors.mkString(", "))
println("Sorted: " + kept.sorted.map(s => s.item).mkString(", "))
groups.keys.toList.sorted.foreach(k => println(k + ": " + groups(k).map(s => s.amount).sum.toString))
println("Total: " + kept.map(s => s.amount).sum.toString)
Key points
- Validation: checking each record and collecting every problem found.
- Accumulation: gathering all errors in one list instead of stopping at the first problem.
- Either: a type that holds a result on the right side or a list of errors on the left side.
- Grouping: collecting records that share a key into separate lists.
- Ordering: a given rule that decides the sort order for a type.
In this example, the raw rows parse into sales or error lists. All errors print together, the kept sales sort largest first, and each category totals separately. The final total is 120.
Note
Scala prints a List with a comma and a space between items.
Try it Yourself
Edit the code, then press Run.
Running is simulated · Scastie
Exercise
A record has an empty item and a bad amount. What does validate return?