Avoid setting job names with updates, ignore zero accumulators

iht · Dec 21, 2024 · 5ff04c1 · 5ff04c1
1 parent f97a5ca
commit 5ff04c1
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 5 deletions.
diff --git a/src/main/java/com/google/cloud/pso/RunPipeline.java b/src/main/java/com/google/cloud/pso/RunPipeline.java
@@ -19,6 +19,7 @@
 import com.google.cloud.pso.options.TaxiSessionsOptions;
 import com.google.cloud.pso.pipelines.TaxiSessionsPipeline;
 import java.time.Instant;
+import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 
@@ -43,11 +44,11 @@ public static void main(String[] args) {
     }
 
     if (p != null) {
-      String jobName = options.getJobName();
-      // Set name only if not empty.
-      // If jobName is already set, it might be because we are updating the job.
-      if (jobName == null || jobName.isEmpty()) jobName = getJobName(pipelineToRun);
-      p.getOptions().setJobName(jobName);
+      DataflowPipelineOptions dataflow = options.as(DataflowPipelineOptions.class);
+      // Set name only if we are not updating a Dataflow pipeline
+      if (!dataflow.isUpdate()) {
+        dataflow.setJobName(getJobName(pipelineToRun));
+      }
       p.run().waitUntilFinish();
     } else {
       System.out.println("Unrecognized pipeline type " + pipelineToRun);

diff --git a/src/main/java/com/google/cloud/pso/transforms/Session.java b/src/main/java/com/google/cloud/pso/transforms/Session.java
@@ -140,6 +140,10 @@ public RideAccumulator mergeAccumulators(Iterable<RideAccumulator> accumulators)
       int count = 0;
 
       for (RideAccumulator accumulator : accumulators) {
+        if (accumulator.countEvents == 0) {
+          // Ignoring empty accumlator ("zeros")
+          continue;
+        }
         count += accumulator.countEvents;
         if (begin == null || accumulator.beginTimestamp.isBefore(begin)) {
           begin = accumulator.beginTimestamp;