Remove spaces in transform names

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

Hide file tree

Showing 2 changed files with 7 additions and 7 deletions.
diff --git a/src/main/java/com/google/cloud/pso/pipelines/TaxiSessionsPipeline.java b/src/main/java/com/google/cloud/pso/pipelines/TaxiSessionsPipeline.java
@@ -61,12 +61,12 @@ public static Pipeline createPipeline(TaxiSessionsOptions opts) {
     // triggers, etc).
     PCollection<String> rides =
         pipeline.apply(
-            "Read rides",
+            "ReadRides",
             PubsubIO.readStrings()
                 .fromSubscription(opts.getRideEventsSubscription())
                 .withTimestampAttribute("ts"));
 
-    PCollectionTuple parsed = rides.apply("Parse JSON", Parser.TaxiEventParser.parseJson());
+    PCollectionTuple parsed = rides.apply("ParseJSON", Parser.TaxiEventParser.parseJson());
     PCollection<RideEvent> rideEvents = parsed.get(Parser.TaxiEventParser.TAXI_EVENT_TAG);
     PCollection<ParsingError> parsingErrors = parsed.get(Parser.TaxiEventParser.ERROR_TAG);
 
@@ -79,7 +79,7 @@ public static Pipeline createPipeline(TaxiSessionsOptions opts) {
                 .build());
 
     parsingErrors.apply(
-        "Append errors to BQ",
+        "AppendErrorsToBQ",
         BigQueryIO.<ParsingError>write()
             .to(errorsTable)
             .useBeamSchema()
@@ -100,7 +100,7 @@ public static Pipeline createPipeline(TaxiSessionsOptions opts) {
         (ValueInSingleWindow<RideSession> s) -> rideSessionTDestination(s, project, dataset);
 
     rideSessions.apply(
-        "Upsert sessions in BQ",
+        "UpsertSessionsBQ",
         BigQueryIO.<RideSession>write()
             .to(tableFunc)
             .useBeamSchema()

diff --git a/src/main/java/com/google/cloud/pso/transforms/Session.java b/src/main/java/com/google/cloud/pso/transforms/Session.java
@@ -72,7 +72,7 @@ public PCollection<RideSession> expand(PCollection<RideEvent> rideEvents) {
       // Add keys to rideEvents
       PCollection<KV<String, RideEvent>> withKeys =
           rideEvents.apply(
-              "Add keys", WithKeys.of(RideEvent::getRideId).withKeyType(TypeDescriptors.strings()));
+              "AddKeys", WithKeys.of(RideEvent::getRideId).withKeyType(TypeDescriptors.strings()));
 
       // Use a late trigger only if late data wait is not zero
       Trigger sessionTrigger;
@@ -99,9 +99,9 @@ public PCollection<RideSession> expand(PCollection<RideEvent> rideEvents) {
                   .withAllowedLateness(Duration.standardSeconds(lateDataWaitSeconds())));
 
       PCollection<KV<String, RideAccumulator>> accumulated =
-          sessions.apply("Combine events", Combine.perKey(new SessionPropertiesCombinerFn()));
+          sessions.apply("CombineEvents", Combine.perKey(new SessionPropertiesCombinerFn()));
 
-      return accumulated.apply("Session properties", ParDo.of(new SessionPropertiesDoFn()));
+      return accumulated.apply("SessionProperties", ParDo.of(new SessionPropertiesDoFn()));
     }
   }