Ardagan commented on a change in pull request #11477:
URL: https://github.com/apache/beam/pull/11477#discussion_r418673674
##########
File path:
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/PeriodicSequence.java
##########
@@ -21,33 +21,69 @@
import static
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
import static
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkState;
-import java.util.List;
+import java.util.Objects;
import javax.annotation.Nullable;
import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.schemas.JavaFieldSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
-import org.apache.beam.sdk.transforms.splittabledofn.Sizes;
import org.apache.beam.sdk.transforms.splittabledofn.SplitResult;
import org.apache.beam.sdk.values.PCollection;
import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
import org.joda.time.Duration;
import org.joda.time.Instant;
/**
- * A {@link PTransform} which generates a sequence of timestamped elements at
given interval in
- * runtime.
+ * A {@link PTransform} which generates a sequence of timestamped elements at
given runtime
+ * interval.
*
- * <p>Receives a PCollection<List<Long>> where each element triggers the
generation of sequence and
- * has following elements: 0: first element timestamp 1: last element
timestamp 2: interval
+ * <p>Transform will not output elements prior to target time. Transform can
output elements at any
+ * time after target time.
*
- * <p>All elements that have timestamp in the past will be output right away.
Elements that have
- * timestamp in the future will be delayed.
- *
- * <p>Transform will not output elements prior to target timestamp. Transform
can output elements at
- * any time after target timestamp.
+ * <p>Multiple elements can be output at given moment if their timestamp is
earlier than current
+ * time.
*/
@Experimental(Experimental.Kind.SPLITTABLE_DO_FN)
-public class PeriodicSequence extends PTransform<PCollection<List<Long>>,
PCollection<Instant>> {
+public class PeriodicSequence
+ extends PTransform<PCollection<PeriodicSequence.SequenceDefinition>,
PCollection<Instant>> {
+
+ @DefaultSchema(JavaFieldSchema.class)
+ public static class SequenceDefinition {
+ public Instant first;
+ public Instant last;
+ public Long durationMilliSec;
+
+ public SequenceDefinition() {}
+
+ public SequenceDefinition(Instant first, Instant last, Duration duration) {
+ this.first = first;
+ this.last = last;
+ this.durationMilliSec = duration.getMillis();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+
+ if (obj == null || obj.getClass() != this.getClass()) {
+ return false;
+ }
+
+ SequenceDefinition src = (SequenceDefinition) obj;
+ return src.first.equals(this.first)
+ && src.last.equals(this.last)
+ && src.durationMilliSec.equals(this.durationMilliSec);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = Objects.hash(first, last, durationMilliSec);
+ return result;
+ }
Review comment:
I tried to use it, but didn't manage to get AutoValue work. It failed to
properly detect constructor for generated class. I tried to debug it, but it
would be better to handle in separate PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
