Scala + Tools

import org.apache.spark.sql.{SparkSession, functions as F}

case class Order(id: Long, country: String, amount: Double)

object Revenue {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("revenue-by-country")
      .master("local[*]")          // drop this line when using spark-submit
      .getOrCreate()
    import spark.implicits._

    val orders = spark.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv("data/orders.csv")
      .as[Order]                  // typed Dataset[Order]

    val revenue = orders
      .filter(_.amount > 0)       // typed lambda
      .groupBy($"country")
      .agg(F.round(F.sum($"amount"), 2).as("revenue"), F.count("*").as("orders"))
      .orderBy(F.desc("revenue"))

    revenue.show()                // action: the job runs here
    revenue.write.mode("overwrite").parquet("out/revenue")
    spark.stop()
  }
}

import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import java.time.Duration
import java.util.Properties
import scala.jdk.CollectionConverters.*
import scala.util.Using

val common = Map("bootstrap.servers" -> "localhost:9092")

def props(m: Map[String, String]): Properties =
  val p = Properties()
  m.foreach((k, v) => p.put(k, v))
  p

@main def produce(): Unit =
  val cfg = common ++ Map(
    "key.serializer"   -> "org.apache.kafka.common.serialization.StringSerializer",
    "value.serializer" -> "org.apache.kafka.common.serialization.StringSerializer",
    "acks"             -> "all",
  )
  Using.resource(KafkaProducer[String, String](props(cfg))) { producer =>
    producer.send(ProducerRecord("orders", "order-42", """{"id":42,"status":"paid"}""")).get()
  }

@main def consume(): Unit =
  val cfg = common ++ Map(
    "group.id"           -> "billing",
    "enable.auto.commit" -> "false",
    "auto.offset.reset"  -> "earliest",
    "key.deserializer"   -> "org.apache.kafka.common.serialization.StringDeserializer",
    "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
  )
  Using.resource(KafkaConsumer[String, String](props(cfg))) { consumer =>
    consumer.subscribe(List("orders").asJava)
    while true do
      val records = consumer.poll(Duration.ofMillis(500)).asScala
      records.foreach(r => println(s"${r.key} @ partition ${r.partition} offset ${r.offset}: ${r.value}"))
      if records.nonEmpty then consumer.commitSync()
  }

import cats.effect.{IO, IOApp}
import doobie.*
import doobie.implicits.*

case class Customer(id: Long, email: String)

object CustomersApp extends IOApp.Simple:
  val xa = Transactor.fromDriverManager[IO](
    driver = "org.postgresql.Driver",
    url = "jdbc:postgresql://localhost:5432/shop",
    user = "shop",
    password = sys.env.getOrElse("DB_PASSWORD", ""),
    logHandler = None,
  )

  def byDomain(domain: String): ConnectionIO[List[Customer]] =
    sql"SELECT id, email FROM customers WHERE email LIKE ${"%@" + domain} ORDER BY id"
      .query[Customer]
      .to[List]

  def run: IO[Unit] =
    byDomain("example.com").transact(xa).flatMap(cs => IO.println(cs))

// List(Customer(1,[email protected]), Customer(4,[email protected]))

import redis.clients.jedis.JedisPool
import scala.util.Using

class PriceCache(pool: JedisPool, loadFromDb: String => BigDecimal):
  def price(sku: String): BigDecimal =
    Using.resource(pool.getResource) { redis =>
      Option(redis.get(s"price:$sku")) match
        case Some(cached) => BigDecimal(cached)          // hit
        case None =>                                      // miss: load and store
          val fresh = loadFromDb(sku)
          redis.setex(s"price:$sku", 300, fresh.toString)
          fresh
    }

// val cache = PriceCache(JedisPool("localhost", 6379), sku => slowQuery(sku))
// cache.price("SKU-1")  // first call: database; next 5 minutes: Redis

import software.amazon.awssdk.core.sync.RequestBody
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.services.s3.S3Client
import software.amazon.awssdk.services.s3.model.{ListObjectsV2Request, PutObjectRequest}
import scala.jdk.CollectionConverters.*
import scala.util.Using

@main def s3Demo(): Unit =
  Using.resource(S3Client.builder().region(Region.AP_SOUTH_1).build()) { s3 =>
    val bucket = "acme-data-lake"

    s3.putObject(
      PutObjectRequest.builder().bucket(bucket).key("raw/orders/2024-06-01.csv").build(),
      RequestBody.fromString("id,country,amount\n1,IN,120.50\n"),
    )

    val listing = s3.listObjectsV2Paginator(
      ListObjectsV2Request.builder().bucket(bucket).prefix("raw/orders/").build()
    )
    for obj <- listing.contents().asScala do
      println(s"${obj.key} ${obj.size} bytes")
  }

# build stage
FROM sbtscala/scala-sbt:eclipse-temurin-21.0.7_6_1.11.6_3.3.6 AS build
WORKDIR /app
COPY build.sbt ./
COPY project ./project
RUN sbt update                      # dependencies cached in their own layer
COPY src ./src
RUN sbt assembly                    # target/scala-3.3.6/orders-assembly-0.1.0.jar

# run stage
FROM eclipse-temurin:21-jre
WORKDIR /app
COPY --from=build /app/target/scala-3.3.6/*-assembly-*.jar app.jar
USER 1000
EXPOSE 8080
ENTRYPOINT ["java", "-XX:MaxRAMPercentage=75", "-jar", "app.jar"]

# docker build -t orders:1.0 .
# docker run -p 8080:8080 -e DB_PASSWORD=... orders:1.0

# .github/workflows/ci.yml
name: ci
on:
  push:
    branches: [main]
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-java@v4
        with:
          distribution: temurin
          java-version: 21
          cache: sbt
      - uses: sbt/setup-sbt@v1
      - name: Format, compile, test
        run: sbt scalafmtCheckAll compile test