pabloem commented on a change in pull request #14531: URL: https://github.com/apache/beam/pull/14531#discussion_r630491759
########## File path: examples/java/src/main/java/org/apache/beam/examples/complete/twitterstreamgenerator/ReadFromTwitterDoFn.java ########## @@ -0,0 +1,212 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.examples.complete.twitterstreamgenerator; + +import java.io.IOException; +import java.io.Serializable; +import java.util.Objects; +import java.util.concurrent.BlockingQueue; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.SerializableCoder; +import org.apache.beam.sdk.io.range.OffsetRange; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.splittabledofn.ManualWatermarkEstimator; +import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker; +import org.apache.beam.sdk.transforms.splittabledofn.SplitResult; +import org.apache.beam.sdk.transforms.splittabledofn.WatermarkEstimators; +import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.DateTime; +import org.joda.time.Duration; +import org.joda.time.Instant; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import twitter4j.Status; + +/** Splittable dofn that read live data off twitter. * */ [email protected] +final class ReadFromTwitterDoFn extends DoFn<TwitterConfig, String> { + + private final DateTime startTime; + + ReadFromTwitterDoFn() { + this.startTime = new DateTime(); + } + /* Logger for class.*/ + private static final Logger LOG = LoggerFactory.getLogger(ReadFromTwitterDoFn.class); + + static class OffsetHolder implements Serializable { + public final @Nullable TwitterConfig twitterConfig; + public final @Nullable Long fetchedRecords; + + OffsetHolder(@Nullable TwitterConfig twitterConfig, @Nullable Long fetchedRecords) { + this.twitterConfig = twitterConfig; + this.fetchedRecords = fetchedRecords; + } + + @Override + public boolean equals(@Nullable Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + OffsetHolder that = (OffsetHolder) o; + return Objects.equals(twitterConfig, that.twitterConfig) + && Objects.equals(fetchedRecords, that.fetchedRecords); + } + + @Override + public int hashCode() { + return Objects.hash(twitterConfig, fetchedRecords); + } + } + + static class OffsetTracker extends RestrictionTracker<OffsetHolder, TwitterConfig> + implements Serializable { + private OffsetHolder restriction; + private final DateTime startTime; + + OffsetTracker(OffsetHolder holder, DateTime startTime) { + this.restriction = holder; + this.startTime = startTime; + } + + @Override + public boolean tryClaim(TwitterConfig twitterConfig) { + LOG.info( + "-------------- Claiming " + + twitterConfig.hashCode() + + " used to have: " + + restriction.fetchedRecords); + long fetchedRecords = + this.restriction == null || this.restriction.fetchedRecords == null + ? 0 + : this.restriction.fetchedRecords + 1; + long elapsedTime = System.currentTimeMillis() - startTime.getMillis(); + final long millis = 60 * 1000; + LOG.info( + "-------------- Time running: {} / {}", + elapsedTime, + (twitterConfig.getMinutesToRun() * millis)); + if (fetchedRecords > twitterConfig.getTweetsCount() + || elapsedTime > twitterConfig.getMinutesToRun() * millis) { + return false; + } + this.restriction = new OffsetHolder(twitterConfig, fetchedRecords); + return true; + } + + @Override + public OffsetHolder currentRestriction() { + return restriction; + } + + @Override + public SplitResult<OffsetHolder> trySplit(double fractionOfRemainder) { + LOG.info("-------------- Trying to split: fractionOfRemainder=" + fractionOfRemainder); + return SplitResult.of(new OffsetHolder(null, 0L), restriction); + } + + @Override + public void checkDone() throws IllegalStateException {} + + @Override + public IsBounded isBounded() { + return IsBounded.UNBOUNDED; + } + } + + @GetInitialWatermarkEstimatorState + public Instant getInitialWatermarkEstimatorState(@Timestamp Instant currentElementTimestamp) { + return currentElementTimestamp; + } + + private static Instant ensureTimestampWithinBounds(Instant timestamp) { + if (timestamp.isBefore(BoundedWindow.TIMESTAMP_MIN_VALUE)) { + timestamp = BoundedWindow.TIMESTAMP_MIN_VALUE; + } else if (timestamp.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE)) { + timestamp = BoundedWindow.TIMESTAMP_MAX_VALUE; + } + return timestamp; + } + + @NewWatermarkEstimator + public WatermarkEstimators.Manual newWatermarkEstimator( + @WatermarkEstimatorState Instant watermarkEstimatorState) { + return new WatermarkEstimators.Manual(ensureTimestampWithinBounds(watermarkEstimatorState)); + } + + @DoFn.GetInitialRestriction + public OffsetHolder getInitialRestriction(@Element TwitterConfig twitterConfig) + throws IOException { + return new OffsetHolder(null, 0L); + } + + @DoFn.NewTracker + public RestrictionTracker<OffsetHolder, TwitterConfig> newTracker( + @Element TwitterConfig twitterConfig, @DoFn.Restriction OffsetHolder restriction) { + return new OffsetTracker(restriction, startTime); + } + + @GetRestrictionCoder + public Coder<OffsetHolder> getRestrictionCoder() { + return SerializableCoder.of(OffsetHolder.class); + } + + @DoFn.ProcessElement + public DoFn.ProcessContinuation processElement( + @Element TwitterConfig twitterConfig, + DoFn.OutputReceiver<String> out, + RestrictionTracker<OffsetRange, TwitterConfig> tracker, + ManualWatermarkEstimator<Instant> watermarkEstimator) { + LOG.info("In Read From Twitter Do Fn"); Review comment: ditto ########## File path: examples/java/src/main/java/org/apache/beam/examples/complete/twitterstreamgenerator/ReadFromTwitterDoFn.java ########## @@ -0,0 +1,212 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.examples.complete.twitterstreamgenerator; + +import java.io.IOException; +import java.io.Serializable; +import java.util.Objects; +import java.util.concurrent.BlockingQueue; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.SerializableCoder; +import org.apache.beam.sdk.io.range.OffsetRange; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.splittabledofn.ManualWatermarkEstimator; +import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker; +import org.apache.beam.sdk.transforms.splittabledofn.SplitResult; +import org.apache.beam.sdk.transforms.splittabledofn.WatermarkEstimators; +import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.DateTime; +import org.joda.time.Duration; +import org.joda.time.Instant; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import twitter4j.Status; + +/** Splittable dofn that read live data off twitter. * */ [email protected] +final class ReadFromTwitterDoFn extends DoFn<TwitterConfig, String> { + + private final DateTime startTime; + + ReadFromTwitterDoFn() { + this.startTime = new DateTime(); + } + /* Logger for class.*/ + private static final Logger LOG = LoggerFactory.getLogger(ReadFromTwitterDoFn.class); + + static class OffsetHolder implements Serializable { + public final @Nullable TwitterConfig twitterConfig; + public final @Nullable Long fetchedRecords; + + OffsetHolder(@Nullable TwitterConfig twitterConfig, @Nullable Long fetchedRecords) { + this.twitterConfig = twitterConfig; + this.fetchedRecords = fetchedRecords; + } + + @Override + public boolean equals(@Nullable Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + OffsetHolder that = (OffsetHolder) o; + return Objects.equals(twitterConfig, that.twitterConfig) + && Objects.equals(fetchedRecords, that.fetchedRecords); + } + + @Override + public int hashCode() { + return Objects.hash(twitterConfig, fetchedRecords); + } + } + + static class OffsetTracker extends RestrictionTracker<OffsetHolder, TwitterConfig> + implements Serializable { + private OffsetHolder restriction; + private final DateTime startTime; + + OffsetTracker(OffsetHolder holder, DateTime startTime) { + this.restriction = holder; + this.startTime = startTime; + } + + @Override + public boolean tryClaim(TwitterConfig twitterConfig) { + LOG.info( + "-------------- Claiming " + + twitterConfig.hashCode() + + " used to have: " + + restriction.fetchedRecords); + long fetchedRecords = + this.restriction == null || this.restriction.fetchedRecords == null + ? 0 + : this.restriction.fetchedRecords + 1; + long elapsedTime = System.currentTimeMillis() - startTime.getMillis(); + final long millis = 60 * 1000; + LOG.info( Review comment: ditto ########## File path: examples/java/src/main/java/org/apache/beam/examples/complete/twitterstreamgenerator/ReadFromTwitterDoFn.java ########## @@ -0,0 +1,212 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.examples.complete.twitterstreamgenerator; + +import java.io.IOException; +import java.io.Serializable; +import java.util.Objects; +import java.util.concurrent.BlockingQueue; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.SerializableCoder; +import org.apache.beam.sdk.io.range.OffsetRange; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.splittabledofn.ManualWatermarkEstimator; +import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker; +import org.apache.beam.sdk.transforms.splittabledofn.SplitResult; +import org.apache.beam.sdk.transforms.splittabledofn.WatermarkEstimators; +import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.DateTime; +import org.joda.time.Duration; +import org.joda.time.Instant; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import twitter4j.Status; + +/** Splittable dofn that read live data off twitter. * */ [email protected] +final class ReadFromTwitterDoFn extends DoFn<TwitterConfig, String> { + + private final DateTime startTime; + + ReadFromTwitterDoFn() { + this.startTime = new DateTime(); + } + /* Logger for class.*/ + private static final Logger LOG = LoggerFactory.getLogger(ReadFromTwitterDoFn.class); + + static class OffsetHolder implements Serializable { + public final @Nullable TwitterConfig twitterConfig; + public final @Nullable Long fetchedRecords; + + OffsetHolder(@Nullable TwitterConfig twitterConfig, @Nullable Long fetchedRecords) { + this.twitterConfig = twitterConfig; + this.fetchedRecords = fetchedRecords; + } + + @Override + public boolean equals(@Nullable Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + OffsetHolder that = (OffsetHolder) o; + return Objects.equals(twitterConfig, that.twitterConfig) + && Objects.equals(fetchedRecords, that.fetchedRecords); + } + + @Override + public int hashCode() { + return Objects.hash(twitterConfig, fetchedRecords); + } + } + + static class OffsetTracker extends RestrictionTracker<OffsetHolder, TwitterConfig> + implements Serializable { + private OffsetHolder restriction; + private final DateTime startTime; + + OffsetTracker(OffsetHolder holder, DateTime startTime) { + this.restriction = holder; + this.startTime = startTime; + } + + @Override + public boolean tryClaim(TwitterConfig twitterConfig) { + LOG.info( + "-------------- Claiming " Review comment: perhaps remove the debug logging or make it `debug` level ########## File path: examples/java/src/main/java/org/apache/beam/examples/complete/twitterstreamgenerator/TwitterIO.java ########## @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.examples.complete.twitterstreamgenerator; + +import java.util.ArrayList; +import java.util.List; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.PTransform; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.values.PBegin; +import org.apache.beam.sdk.values.PCollection; + +/** + * An unbounded source for <a + * href="https://developer.twitter.com/en/docs/tutorials/consuming-streaming-data";>twitter</a> + * stream. PTransforms for streaming live tweets from twitter. Reading from Twitter is supported by + * read() + * + * <p>Standard Twitter API can be read using a list of Twitter Config + * readStandardStream(List<TwitterConfig>) + * + * <p>We allow multiple Twitter configurations to demonstrate how multiple twitter streams can be + * combined in a single pipeline. + * + * <p>TwitterIO.readStandardStream( Arrays.asList( new TwitterConfig .Builder() .setKey("") + * .setSecret("") .setToken("") .setTokenSecret("") .setFilters(Arrays.asList("", "")) + * .setLanguage("en") .setTweetsCount(10L) .setMinutesToRun(1) .build())) + */ +public class TwitterIO { + + /** + * Initializes the stream by converting input to a Twitter connection configuration. + * + * @param twitterConfigs list of twitter config + * @return PTransform of statuses + */ + public static PTransform<PBegin, PCollection<String>> readStandardStream( + List<TwitterConfig> twitterConfigs) { Review comment: will this work properly with multiple configs? have you tested that? let's provide a constructor that takes a single config, so that users don't need to create a whole list for a source that reads only one config. Thoughts? ########## File path: examples/java/src/main/java/org/apache/beam/examples/complete/twitterstreamgenerator/TwitterIO.java ########## @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.examples.complete.twitterstreamgenerator; + +import java.util.ArrayList; +import java.util.List; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.PTransform; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.values.PBegin; +import org.apache.beam.sdk.values.PCollection; + +/** + * An unbounded source for <a + * href="https://developer.twitter.com/en/docs/tutorials/consuming-streaming-data";>twitter</a> + * stream. PTransforms for streaming live tweets from twitter. Reading from Twitter is supported by + * read() + * + * <p>Standard Twitter API can be read using a list of Twitter Config + * readStandardStream(List<TwitterConfig>) + * + * <p>We allow multiple Twitter configurations to demonstrate how multiple twitter streams can be + * combined in a single pipeline. + * + * <p>TwitterIO.readStandardStream( Arrays.asList( new TwitterConfig .Builder() .setKey("") + * .setSecret("") .setToken("") .setTokenSecret("") .setFilters(Arrays.asList("", "")) + * .setLanguage("en") .setTweetsCount(10L) .setMinutesToRun(1) .build())) + */ Review comment: check how we add code-type javadoc in BQIO: https://github.com/apache/beam/blob/v2.24.0/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java#L250-L267 please add it so it's nicely formatted like htat -- 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]
