Bump to Scio v0.14.10 and Beam 2.61

iht · iht · commit f08e6896fccc · 2025-01-16T22:46:10.000+01:00
diff --git a/build.sbt b/build.sbt
@@ -1,18 +1,63 @@
 import sbt._
 import Keys._
-val scioVersion = "0.10.4"
-val beamVersion = "2.30.0"
-lazy val commonSettings = Def.settings(
-  organization := "dev.herraiz",
+import com.here.bom.Bom
+
+val scioVersion = "0.14.10"
+val beamVersion = "2.61.0"
+
+val guavaVersion = "33.1.0-jre"
+val jacksonVersion = "2.15.4"
+val magnolifyVersion = "0.7.4"
+val nettyVersion = "4.1.100.Final"
+val slf4jVersion = "1.7.30"
+val gcpLibrariesVersion = "26.45.0"
+
+lazy val gcpBom = Bom(
+  "com.google.cloud" % "libraries-bom" % gcpLibrariesVersion
+)
+lazy val beamBom = Bom("org.apache.beam" % "beam-sdks-java-bom" % beamVersion)
+lazy val guavaBom = Bom("com.google.guava" % "guava-bom" % guavaVersion)
+lazy val jacksonBom = Bom(
+  "com.fasterxml.jackson" % "jackson-bom" % jacksonVersion
+)
+lazy val magnolifyBom = Bom("com.spotify" % "magnolify-bom" % magnolifyVersion)
+lazy val nettyBom = Bom("io.netty" % "netty-bom" % nettyVersion)
+lazy val scioBom = Bom("com.spotify" % "scio-bom" % scioVersion)
+
+val bomSettings = Def.settings(
+  gcpBom,
+  beamBom,
+  guavaBom,
+  jacksonBom,
+  magnolifyBom,
+  nettyBom,
+  dependencyOverrides ++=
+    gcpBom.key.value.bomDependencies ++
+      beamBom.key.value.bomDependencies ++
+      guavaBom.key.value.bomDependencies ++
+      jacksonBom.key.value.bomDependencies ++
+      magnolifyBom.key.value.bomDependencies ++
+      nettyBom.key.value.bomDependencies
+)
+
+lazy val commonSettings = bomSettings ++ Def.settings(
+  organization := "example",
   // Semantic versioning http://semver.org/
   version := "0.1.0-SNAPSHOT",
-  scalaVersion := "2.13.16",
-  scalacOptions ++= Seq("-target:jvm-1.8",
-                        "-deprecation",
-                        "-feature",
-                        "-unchecked",
-                        "-Ymacro-annotations"),
-  javacOptions ++= Seq("-source", "1.8", "-target", "1.8")
+  scalaVersion := "2.13.15",
+  scalacOptions ++= Seq(
+    "-release",
+    "11",
+    "-deprecation",
+    "-feature",
+    "-unchecked",
+    "-Ymacro-annotations"
+  ),
+  javacOptions ++= Seq("--release", "11"),
+  // add extra resolved and remove exclude if you need kafka
+  // resolvers += "confluent" at "https://packages.confluent.io/maven/",
+  excludeDependencies += "org.apache.beam" % "beam-sdks-java-io-kafka",
+  excludeDependencies += "com.github.luben" % "zstd-jni"
 )
 
 lazy val root: Project = project
@@ -22,19 +67,21 @@ lazy val root: Project = project
     name := "scio-scala-workshop-beam-summit",
     description := "scio-scala-workshop-beam-summit",
     publish / skip := true,
-    run / classLoaderLayeringStrategy := ClassLoaderLayeringStrategy.Flat,
+    fork := true,
+    run / outputStrategy := Some(OutputStrategy.StdoutOutput),
     libraryDependencies ++= Seq(
       "com.spotify" %% "scio-core" % scioVersion,
       "com.spotify" %% "scio-google-cloud-platform" % scioVersion,
       "com.spotify" %% "scio-extra" % scioVersion,
       "com.spotify" %% "scio-test" % scioVersion % Test,
-      "org.apache.beam" % "beam-runners-direct-java" % beamVersion,
-      "org.apache.beam" % "beam-runners-google-cloud-dataflow-java" % beamVersion,
-      "org.slf4j" % "slf4j-simple" % "2.0.16",
-      "com.google.http-client" % "google-http-client-apache-v2" % "1.45.3"
+      "org.slf4j" % "slf4j-api" % slf4jVersion,
+      "org.apache.beam" % "beam-runners-google-cloud-dataflow-java" % beamVersion % Runtime,
+      "org.apache.beam" % "beam-runners-direct-java" % beamVersion % Test,
+      "com.spotify" %% "scio-test" % scioVersion % Test,
+      "org.slf4j" % "slf4j-simple" % slf4jVersion % Test,
+      "com.github.luben" % "zstd-jni" % "1.5.2-2"
     )
   )
-  .enablePlugins(JavaAppPackaging)
 
 lazy val repl: Project = project
   .in(file(".repl"))
@@ -46,8 +93,9 @@ lazy val repl: Project = project
       "com.spotify" %% "scio-repl" % scioVersion
     ),
     Compile / mainClass := Some("com.spotify.scio.repl.ScioShell"),
-    publish / skip := true
+    publish / skip := true,
+    fork := false
   )
   .dependsOn(root)
 
-resolvers += "confluent" at "https://packages.confluent.io/maven/"
+ThisBuild / versionPolicyIntention := Compatibility.BinaryAndSourceCompatible
diff --git a/project/plugins.sbt b/project/plugins.sbt
@@ -1 +1,3 @@
-addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.8.1")
+addDependencyTreePlugin
+addSbtPlugin("com.here.platform" % "sbt-bom" % "1.0.17")
+addSbtPlugin("ch.epfl.scala" % "sbt-version-policy" % "3.2.1")
diff --git a/src/main/scala/dev/herraiz/TaxiSessionsPipeline.scala b/src/main/scala/dev/herraiz/TaxiSessionsPipeline.scala
@@ -16,13 +16,16 @@
 
 package dev.herraiz
 
+import io.circe.Error
+import com.spotify.scio._
 import com.spotify.scio.bigquery._
 import com.spotify.scio.pubsub._
-import com.spotify.scio.values.{SCollection, WindowOptions}
-import com.spotify.scio.{Args, ContextAndArgs, ScioContext, streaming}
+import com.spotify.scio.values._
 import dev.herraiz.data.DataTypes._
-import io.circe
-import org.apache.beam.sdk.transforms.windowing.{AfterProcessingTime, AfterWatermark}
+import org.apache.beam.sdk.transforms.windowing.{
+  AfterProcessingTime,
+  AfterWatermark
+}
 import org.joda.time.Duration
 
 object TaxiSessionsPipeline {
@@ -40,67 +43,96 @@ object TaxiSessionsPipeline {
     val accumTable = opts("accum-table")
 
     val messages: SCollection[String] = getMessagesFromPubSub(pubsubTopic)
-    val (rides: SCollection[PointTaxiRide], writableErrors: SCollection[JsonError]) = parseJSONStrings(messages)
-
-    rides.saveAsBigQueryTable(Table.Spec(goodTable), WRITE_APPEND, CREATE_IF_NEEDED)
-    writableErrors.saveAsBigQueryTable(Table.Spec(badTable), WRITE_APPEND, CREATE_IF_NEEDED)
+    val (
+      rides: SCollection[PointTaxiRide],
+      writableErrors: SCollection[JsonError]
+    ) = parseJSONStrings(messages)
+
+    rides
+      .saveAsTypedBigQueryTable(
+        Table.Spec(goodTable)
+      )
+
+    writableErrors.saveAsTypedBigQueryTable(
+      Table.Spec(badTable)
+    )
 
     // Group by session with a max duration of 5 mins between events
     // Window options
     val wopts: WindowOptions = customWindowOptions
     val groupRides = groupRidesByKey(rides.map(_.toTaxiRide), wopts)
-    groupRides.saveAsBigQueryTable(Table.Spec(accumTable), WRITE_APPEND, CREATE_IF_NEEDED)
+    groupRides.saveAsTypedBigQueryTable(
+      Table.Spec(accumTable)
+    )
 
     sc.run
   }
 
   def customWindowOptions: WindowOptions =
     WindowOptions(
-      trigger = AfterWatermark.pastEndOfWindow()
-        .withEarlyFirings(AfterProcessingTime
-          .pastFirstElementInPane
-          .plusDelayOf(Duration.standardSeconds(EARLY_RESULT)))
-        .withLateFirings(AfterProcessingTime
-          .pastFirstElementInPane()
-          .plusDelayOf(Duration.standardSeconds(LATENESS))),
+      trigger = AfterWatermark
+        .pastEndOfWindow()
+        .withEarlyFirings(
+          AfterProcessingTime.pastFirstElementInPane
+            .plusDelayOf(Duration.standardSeconds(EARLY_RESULT))
+        )
+        .withLateFirings(
+          AfterProcessingTime
+            .pastFirstElementInPane()
+            .plusDelayOf(Duration.standardSeconds(LATENESS))
+        ),
       accumulationMode = streaming.ACCUMULATING_FIRED_PANES,
       allowedLateness = Duration.standardSeconds(LATENESS)
     )
 
-  def getMessagesFromPubSub(pubsubTopic: String)(implicit sc: ScioContext): SCollection[String] = {
-    val pubsubRead: PubsubIO[String] = PubsubIO.string(pubsubTopic, timestampAttribute = "ts")
+  def getMessagesFromPubSub(
+      pubsubTopic: String
+  )(implicit sc: ScioContext): SCollection[String] = {
+    val pubsubRead: PubsubIO[String] =
+      PubsubIO.string(pubsubTopic, timestampAttribute = "ts")
     val pubsubParams: PubsubIO.ReadParam = PubsubIO.ReadParam(PubsubIO.Topic)
 
     /*_*/
     sc.read(pubsubRead)(pubsubParams) /*_*/
   }
 
-  def parseJSONStrings(messages: SCollection[String]):
-  (SCollection[PointTaxiRide], SCollection[JsonError]) = {
-    val jsons: SCollection[Either[circe.Error, PointTaxiRide]] = messages.map { s: String => json2TaxiRide(s) }
-
-    val errorsEither :: pointsEither :: Nil = jsons.partition(2, { e =>
-      e match {
-        case Left(_) => 0
-        case Right(_) => 1
+  def parseJSONStrings(
+      messages: SCollection[String]
+  ): (SCollection[PointTaxiRide], SCollection[JsonError]) = {
+    val jsons: SCollection[Either[Error, PointTaxiRide]] = messages.map {
+      s: String => json2TaxiRide(s)
+    }
+
+    val errorsEither :: pointsEither :: Nil = jsons.partition(
+      2,
+      { e =>
+        e match {
+          case Left(_)  => 0
+          case Right(_) => 1
+        }
       }
-    })
+    )
 
-    val errors: SCollection[circe.Error] = errorsEither.map(_.left.get)
+    val errors: SCollection[Error] = errorsEither.map(_.left.get)
     val points: SCollection[PointTaxiRide] = pointsEither.map(_.right.get)
 
     val jsonErrors: SCollection[JsonError] = errors.map(circeErrorToCustomError)
 
     (points, jsonErrors)
   }
 
-  def groupRidesByKey(rides: SCollection[TaxiRide], wopts: WindowOptions): SCollection[TaxiRide] = {
+  def groupRidesByKey(
+      rides: SCollection[TaxiRide],
+      wopts: WindowOptions
+  ): SCollection[TaxiRide] = {
     val ridesWithKey: SCollection[(String, TaxiRide)] =
       rides.keyBy(_.ride_id)
 
     val afterWindow: SCollection[(String, TaxiRide)] =
-      ridesWithKey.withSessionWindows(Duration.standardSeconds(SESSION_GAP), options = wopts)
-
+      ridesWithKey.withSessionWindows(
+        Duration.standardSeconds(SESSION_GAP),
+        options = wopts
+      )
 
     val agg: SCollection[TaxiRide] = afterWindow.reduceByKey(_ + _).map(_._2)
 
diff --git a/src/main/scala/dev/herraiz/data/TaxiDataTypes.scala b/src/main/scala/dev/herraiz/data/TaxiDataTypes.scala
@@ -16,10 +16,10 @@
 
 package dev.herraiz.data
 
-import com.spotify.scio.bigquery.types.BigQueryType
 import io.circe._
 import io.circe.generic.semiauto._
 import io.circe.parser.decode
+import com.spotify.scio.bigquery.types.BigQueryType
 import org.joda.time.format.{DateTimeFormat, DateTimeFormatter}
 import org.joda.time.{Instant, Interval}
 
@@ -46,27 +46,29 @@ object DataTypes {
 
       ti match {
         case Success(instant: Instant) => Right(instant)
-        case Failure(e: Throwable) => Left(e.getMessage)
+        case Failure(e: Throwable)     => Left(e.getMessage)
       }
     }
 
   // Decoder to produce TaxiRide objects from Json
-  protected lazy implicit val taxiRideDecoder: Decoder[PointTaxiRide] = deriveDecoder[PointTaxiRide]
+  protected lazy implicit val taxiRideDecoder: Decoder[PointTaxiRide] =
+    deriveDecoder[PointTaxiRide]
 
   @BigQueryType.toTable
   case class PointTaxiRide(
-                            ride_id: String,
-                            point_idx: Int,
-                            latitude: Double,
-                            longitude: Double,
-                            timestamp: Instant,
-                            meter_reading: Double,
-                            meter_increment: Double,
-                            ride_status: String,
-                            passenger_count: Int
-                          ) {
+      ride_id: String,
+      point_idx: Int,
+      latitude: Double,
+      longitude: Double,
+      timestamp: Instant,
+      meter_reading: Double,
+      meter_increment: Double,
+      ride_status: String,
+      passenger_count: Int
+  ) {
     def toTaxiRide: TaxiRide =
-      TaxiRide(this.ride_id,
+      TaxiRide(
+        this.ride_id,
         1,
         this.timestamp,
         None,
@@ -78,14 +80,14 @@ object DataTypes {
 
   @BigQueryType.toTable
   case class TaxiRide(
-                       ride_id: String,
-                       n_points: Int,
-                       init: Instant,
-                       finish: Option[Instant],
-                       total_meter: Double,
-                       init_status: String,
-                       finish_status: Option[String]
-                     ) {
+      ride_id: String,
+      n_points: Int,
+      init: Instant,
+      finish: Option[Instant],
+      total_meter: Double,
+      init_status: String,
+      finish_status: Option[String]
+  ) {
     def +(taxiRide: TaxiRide): TaxiRide = {
 
       val (first, second) =
@@ -95,22 +97,28 @@ object DataTypes {
           (this, taxiRide)
         }
 
-
-      val (finishStatus: Option[String], finishInstant: Option[Instant]) = first.finish match {
-        case None =>
-          (Some(second.finish_status.getOrElse(second.init_status)),
-            Some(second.finish.getOrElse(second.init)))
-        case Some(i) =>
-          val interval: Interval = new Interval(first.init, i)
-          val testInstant: Instant = second.finish.getOrElse(second.init)
-          if (interval.contains(testInstant)) {
-            (Some(first.finish_status.getOrElse(first.init_status)),
-              Some(first.finish.getOrElse(first.init)))
-          } else {
-            (Some(second.finish_status.getOrElse(second.init_status)),
-              Some(second.finish.getOrElse(second.init)))
-          }
-      }
+      val (finishStatus: Option[String], finishInstant: Option[Instant]) =
+        first.finish match {
+          case None =>
+            (
+              Some(second.finish_status.getOrElse(second.init_status)),
+              Some(second.finish.getOrElse(second.init))
+            )
+          case Some(i) =>
+            val interval: Interval = new Interval(first.init, i)
+            val testInstant: Instant = second.finish.getOrElse(second.init)
+            if (interval.contains(testInstant)) {
+              (
+                Some(first.finish_status.getOrElse(first.init_status)),
+                Some(first.finish.getOrElse(first.init))
+              )
+            } else {
+              (
+                Some(second.finish_status.getOrElse(second.init_status)),
+                Some(second.finish.getOrElse(second.init))
+              )
+            }
+        }
 
       TaxiRide(
         taxiRide.ride_id,
@@ -127,4 +135,4 @@ object DataTypes {
   @BigQueryType.toTable
   case class JsonError(msg: String)
 
-}
+}
