Add ByKey version

scio-core/src/main/scala/com/spotify/scio/values/PairSCollectionFunctions.scala

@@ -1012,6 +1012,16 @@ class PairSCollectionFunctions[K, V](val self: SCollection[(K, V)]) {
   def minByKey(implicit ord: Ordering[V]): SCollection[(K, V)] =
     this.reduceByKey(ord.min)
 
+  /**
+   * Return latest of values for each key according to its event time, or null if there are no
+   * elements.
+   * @return
+   *   a new SCollection of (key, latest value) pairs
+   * @group per_key
+   */
+  def latestByKey: SCollection[(K, V)] =
+    self.applyPerKey(Latest.perKey[K, V]())(kvToTuple)
+
   /**
    * Merge the values for each key using an associative reduce function. This will also perform the
    * merging locally on each mapper before sending results to a reducer, similarly to a "combiner"

scio-core/src/test/scala/com/spotify/scio/values/PairSCollectionFunctionsTest.scala

@@ -35,6 +35,7 @@ import org.apache.beam.sdk.coders.{
   VarIntCoder,
   ZstdCoder => BZstdCoder
 }
+import org.joda.time.Instant
 
 import scala.collection.mutable
 import scala.jdk.CollectionConverters._
@@ -717,6 +718,17 @@ class PairSCollectionFunctionsTest extends PipelineSpec {
     }
   }
 
+  it should "support latestByKey()" in {
+    runWithContext { sc =>
+      val p = sc
+        .parallelize(Seq(("a", 1L), ("a", 10L), ("b", 2L), ("b", 20L)))
+        .timestampBy { case (_, v) => Instant.ofEpochMilli(v) }
+        .latestByKey
+
+      p should containInAnyOrder(Seq(("a", 10L), ("b", 20L)))
+    }
+  }
+
   it should "support reduceByKey()" in {
     runWithContext { sc =>
       val p = sc