You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ra...@apache.org on 2026/05/13 07:28:05 UTC
(beam) branch master updated: Remove legacy processContext usage across and replace it with argument provider (#38366)
This is an automated email from the ASF dual-hosted git repository.
stankiewicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 79ed7cc166d Remove legacy processContext usage across and replace it with argument provider (#38366)
79ed7cc166d is described below
commit 79ed7cc166d22bf8bd44cdbd80b5694618e413a8
Author: Radosław Stankiewicz <ra...@google.com>
AuthorDate: Wed May 13 09:27:56 2026 +0200
Remove legacy processContext usage across and replace it with argument provider (#38366)
Starting with examples for java and kotlin.
---
.../examples/iceberg/IcebergBatchWriteExample.java | 15 +--
.../beam/examples/SchemaTransformExample.java | 10 +-
.../apache/beam/examples/SqlTransformExample.java | 10 +-
.../beam/examples/ApproximateQuantilesExample.java | 10 +-
.../beam/examples/CoCombineTransformExample.java | 19 ++--
.../apache/beam/examples/CoGroupByKeyExample.java | 10 +-
.../org/apache/beam/examples/CombineExample.java | 10 +-
.../org/apache/beam/examples/CountExample.java | 10 +-
.../apache/beam/examples/CountPerKeyExample.java | 10 +-
.../org/apache/beam/examples/CreateExample.java | 10 +-
.../apache/beam/examples/DebuggingWordCount.java | 15 +--
.../org/apache/beam/examples/DistinctExample.java | 10 +-
.../beam/examples/FlatMapElementsExample.java | 10 +-
.../beam/examples/GroupIntoBatchesExample.java | 10 +-
.../beam/examples/KafkaPassengerCountJson.java | 21 ++--
.../org/apache/beam/examples/KafkaStreaming.java | 33 ++++---
.../apache/beam/examples/KafkaWordCountAvro.java | 11 ++-
.../apache/beam/examples/KafkaWordCountJson.java | 11 ++-
.../java/org/apache/beam/examples/KeysExample.java | 10 +-
.../org/apache/beam/examples/KvSwapExample.java | 10 +-
.../org/apache/beam/examples/LatestExample.java | 10 +-
.../apache/beam/examples/MapElementsExample.java | 10 +-
.../java/org/apache/beam/examples/MaxExample.java | 10 +-
.../org/apache/beam/examples/MaxPerKeyExample.java | 10 +-
.../java/org/apache/beam/examples/MeanExample.java | 10 +-
.../apache/beam/examples/MeanPerKeyExample.java | 10 +-
.../java/org/apache/beam/examples/MinExample.java | 10 +-
.../org/apache/beam/examples/MinPerKeyExample.java | 10 +-
.../org/apache/beam/examples/PartitionExample.java | 10 +-
.../beam/examples/RateLimitedPubSubReader.java | 6 +-
.../apache/beam/examples/RateLimiterSimple.java | 8 +-
.../org/apache/beam/examples/RegexExample.java | 10 +-
.../org/apache/beam/examples/SampleExample.java | 10 +-
.../java/org/apache/beam/examples/SumExample.java | 10 +-
.../org/apache/beam/examples/SumPerKeyExample.java | 10 +-
.../org/apache/beam/examples/ToStringExample.java | 10 +-
.../java/org/apache/beam/examples/TopExample.java | 10 +-
.../org/apache/beam/examples/ValuesExample.java | 10 +-
.../java/org/apache/beam/examples/ViewExample.java | 19 ++--
.../org/apache/beam/examples/WindowExample.java | 10 +-
.../beam/examples/complete/AutoComplete.java | 60 +++++++-----
.../examples/complete/StreamingWordExtract.java | 16 +--
.../org/apache/beam/examples/complete/TfIdf.java | 77 +++++++++------
.../examples/complete/TopWikipediaSessions.java | 32 +++---
.../beam/examples/complete/TrafficMaxLaneFlow.java | 32 +++---
.../beam/examples/complete/TrafficRoutes.java | 48 +++++----
.../transforms/DataProtectors.java | 21 ++--
.../transforms/io/TokenizationBigQueryIO.java | 7 +-
.../transforms/io/TokenizationBigTableIO.java | 6 +-
.../datatokenization/utils/CsvConverters.java | 107 ++++++++++++++-------
.../datatokenization/utils/ErrorConverters.java | 30 +++---
.../beam/examples/complete/game/GameStats.java | 45 +++++----
.../examples/complete/game/HourlyTeamScore.java | 6 +-
.../beam/examples/complete/game/LeaderBoard.java | 18 ++--
.../examples/complete/game/StatefulTeamScore.java | 18 ++--
.../beam/examples/complete/game/UserScore.java | 14 +--
.../complete/game/utils/WriteToBigQuery.java | 22 +++--
.../examples/complete/game/utils/WriteToText.java | 21 ++--
.../game/utils/WriteWindowedToBigQuery.java | 15 ++-
.../cookbook/BigQueryStreamingTornadoes.java | 21 ++--
.../beam/examples/cookbook/BigQueryTornadoes.java | 16 +--
.../examples/cookbook/CombinePerKeyExamples.java | 14 +--
.../beam/examples/cookbook/FilterExamples.java | 31 +++---
.../beam/examples/cookbook/JoinExamples.java | 32 +++---
.../beam/examples/cookbook/MaxPerKeyExamples.java | 17 ++--
.../cookbook/MinimalBigQueryTornadoes.java | 20 ++--
.../beam/examples/cookbook/TriggerExample.java | 47 ++++++---
.../apache/beam/examples/snippets/Snippets.java | 40 +++++---
.../examples/subprocess/ExampleEchoPipeline.java | 10 +-
.../beam/examples/cookbook/TriggerExampleTest.java | 8 +-
.../subprocess/ExampleEchoPipelineTest.java | 10 +-
.../beam/examples/kotlin/DebuggingWordCount.kt | 12 ++-
.../examples/kotlin/cookbook/BigQueryTornadoes.kt | 15 +--
.../kotlin/cookbook/CombinePerKeyExamples.kt | 13 +--
.../examples/kotlin/cookbook/FilterExamples.kt | 30 +++---
.../beam/examples/kotlin/cookbook/JoinExamples.kt | 31 +++---
.../examples/kotlin/cookbook/MaxPerKeyExamples.kt | 15 +--
.../examples/kotlin/cookbook/TriggerExample.kt | 42 +++++---
.../beam/examples/kotlin/snippets/Snippets.kt | 13 +--
79 files changed, 898 insertions(+), 592 deletions(-)
diff --git a/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/IcebergBatchWriteExample.java b/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/IcebergBatchWriteExample.java
index 2a5f85e524e..167d3017c3d 100644
--- a/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/IcebergBatchWriteExample.java
+++ b/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/IcebergBatchWriteExample.java
@@ -28,6 +28,8 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -74,9 +76,8 @@ public class IcebergBatchWriteExample {
static class ExtractBrowserTransactionsFn extends DoFn<Row, KV<String, Long>> {
@ProcessElement
- public void processElement(ProcessContext c) {
- Row row = c.element();
- c.output(
+ public void processElement(@Element Row row, OutputReceiver<KV<String, Long>> receiver) {
+ receiver.output(
KV.of(
Preconditions.checkStateNotNull(row.getString("browser")),
Preconditions.checkStateNotNull(row.getInt64("transactions"))));
@@ -85,13 +86,13 @@ public class IcebergBatchWriteExample {
static class FormatCountsFn extends DoFn<KV<String, Long>, Row> {
@ProcessElement
- public void processElement(ProcessContext c) {
+ public void processElement(@Element KV<String, Long> element, OutputReceiver<Row> receiver) {
Row row =
Row.withSchema(AGGREGATED_SCHEMA)
- .withFieldValue("browser", c.element().getKey())
- .withFieldValue("transaction_count", c.element().getValue())
+ .withFieldValue("browser", element.getKey())
+ .withFieldValue("transaction_count", element.getValue())
.build();
- c.output(row);
+ receiver.output(row);
}
}
diff --git a/examples/java/sql/src/main/java/org/apache/beam/examples/SchemaTransformExample.java b/examples/java/sql/src/main/java/org/apache/beam/examples/SchemaTransformExample.java
index 5b8e4a34cf2..505252e76a6 100644
--- a/examples/java/sql/src/main/java/org/apache/beam/examples/SchemaTransformExample.java
+++ b/examples/java/sql/src/main/java/org/apache/beam/examples/SchemaTransformExample.java
@@ -22,7 +22,7 @@ package org.apache.beam.examples;
// description: Demonstration of Schema transform usage.
// multifile: false
// default_example: false
-// context_line: 60
+// context_line: 65
// categories:
// - Schemas
// - Combiners
@@ -42,6 +42,8 @@ import org.apache.beam.sdk.schemas.transforms.Group;
import org.apache.beam.sdk.schemas.transforms.Select;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Max;
import org.apache.beam.sdk.transforms.Min;
import org.apache.beam.sdk.transforms.ParDo;
@@ -101,9 +103,9 @@ public class SchemaTransformExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/sql/src/main/java/org/apache/beam/examples/SqlTransformExample.java b/examples/java/sql/src/main/java/org/apache/beam/examples/SqlTransformExample.java
index 9c2302c3de2..f2a3a21195f 100644
--- a/examples/java/sql/src/main/java/org/apache/beam/examples/SqlTransformExample.java
+++ b/examples/java/sql/src/main/java/org/apache/beam/examples/SqlTransformExample.java
@@ -22,7 +22,7 @@ package org.apache.beam.examples;
// description: Demonstration of SQL transform usage.
// multifile: false
// default_example: false
-// context_line: 60
+// context_line: 62
// categories:
// - Beam SQL
// - Combiners
@@ -41,6 +41,8 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.Row;
@@ -95,9 +97,9 @@ public class SqlTransformExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/ApproximateQuantilesExample.java b/examples/java/src/main/java/org/apache/beam/examples/ApproximateQuantilesExample.java
index 9e2a96b1eca..daf943484ff 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/ApproximateQuantilesExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/ApproximateQuantilesExample.java
@@ -24,6 +24,8 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.ApproximateQuantiles;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.slf4j.Logger;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of ApproximateQuantiles transform usage.
// multifile: false
// default_example: false
-// context_line: 46
+// context_line: 48
// categories:
// - Core Transforms
// complexity: BASIC
@@ -70,9 +72,9 @@ public class ApproximateQuantilesExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/CoCombineTransformExample.java b/examples/java/src/main/java/org/apache/beam/examples/CoCombineTransformExample.java
index 0ee7d7bcac8..a46ac539aff 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/CoCombineTransformExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/CoCombineTransformExample.java
@@ -22,7 +22,7 @@ package org.apache.beam.examples;
// description: Demonstration of Composed Combine transform usage.
// multifile: false
// default_example: false
-// context_line: 143
+// context_line: 145
// categories:
// - Schemas
// - Combiners
@@ -46,6 +46,8 @@ import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.CombineFns;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Max;
import org.apache.beam.sdk.transforms.Min;
import org.apache.beam.sdk.transforms.ParDo;
@@ -185,13 +187,16 @@ public class CoCombineTransformExample {
new DoFn<
KV<Long, CombineFns.CoCombineResult>, KV<Long, Iterable<KV<String, Long>>>>() {
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- CombineFns.CoCombineResult e = c.element().getValue();
+ public void processElement(
+ @Element KV<Long, CombineFns.CoCombineResult> element,
+ OutputReceiver<KV<Long, Iterable<KV<String, Long>>>> receiver)
+ throws Exception {
+ CombineFns.CoCombineResult e = element.getValue();
ArrayList<KV<String, Long>> o = new ArrayList<KV<String, Long>>();
o.add(KV.of(minTag.getId(), e.get(minTag)));
o.add(KV.of(maxTag.getId(), e.get(maxTag)));
o.add(KV.of(sumTag.getId(), e.get(sumTag)));
- c.output(KV.of(c.element().getKey(), o));
+ receiver.output(KV.of(element.getKey(), o));
}
}));
@@ -210,9 +215,9 @@ public class CoCombineTransformExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/CoGroupByKeyExample.java b/examples/java/src/main/java/org/apache/beam/examples/CoGroupByKeyExample.java
index c77708b5de2..a04c9256a1e 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/CoGroupByKeyExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/CoGroupByKeyExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.join.CoGbkResult;
import org.apache.beam.sdk.transforms.join.CoGroupByKey;
@@ -37,7 +39,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of CoGroupByKey transform usage.
// multifile: false
// default_example: false
-// context_line: 54
+// context_line: 56
// categories:
// - Core Transforms
// - Joins
@@ -84,9 +86,9 @@ public class CoGroupByKeyExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/CombineExample.java b/examples/java/src/main/java/org/apache/beam/examples/CombineExample.java
index 24bed27c236..c4952bfb8d9 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/CombineExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/CombineExample.java
@@ -23,6 +23,8 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.values.PCollection;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Combine transform usage.
// multifile: false
// default_example: false
-// context_line: 47
+// context_line: 49
// categories:
// - Core Transforms
// - Combiners
@@ -68,9 +70,9 @@ public class CombineExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/CountExample.java b/examples/java/src/main/java/org/apache/beam/examples/CountExample.java
index cb0bd0ecf94..80e01b5d6b7 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/CountExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/CountExample.java
@@ -23,6 +23,8 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.slf4j.Logger;
@@ -33,7 +35,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Count transform usage.
// multifile: false
// default_example: false
-// context_line: 45
+// context_line: 47
// categories:
// - Core Transforms
// complexity: BASIC
@@ -63,9 +65,9 @@ public class CountExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/CountPerKeyExample.java b/examples/java/src/main/java/org/apache/beam/examples/CountPerKeyExample.java
index 9bf9bf1ef00..017a736e56f 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/CountPerKeyExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/CountPerKeyExample.java
@@ -23,6 +23,8 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Count.perKey transform usage.
// multifile: false
// default_example: false
-// context_line: 47
+// context_line: 49
// categories:
// - Core Transforms
// complexity: BASIC
@@ -67,9 +69,9 @@ public class CountPerKeyExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/CreateExample.java b/examples/java/src/main/java/org/apache/beam/examples/CreateExample.java
index 5943ffa489d..4052a903147 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/CreateExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/CreateExample.java
@@ -27,6 +27,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
@@ -38,7 +40,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Create transform usage.
// multifile: false
// default_example: false
-// context_line: 51
+// context_line: 53
// categories:
// - Core Transforms
// complexity: BASIC
@@ -79,9 +81,9 @@ public class CreateExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java b/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java
index 7c54e238da3..693aadc7850 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java
@@ -23,7 +23,7 @@ package org.apache.beam.examples;
// filter("Flourish|stomach").
// multifile: false
// pipeline_options: --output output.txt
-// context_line: 180
+// context_line: 182
// categories:
// - Debugging
// - Filtering
@@ -46,6 +46,8 @@ import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
@@ -115,18 +117,19 @@ public class DebuggingWordCount {
private final Counter unmatchedWords = Metrics.counter(FilterTextFn.class, "unmatchedWords");
@ProcessElement
- public void processElement(ProcessContext c) {
- if (filter.matcher(c.element().getKey()).matches()) {
+ public void processElement(
+ @Element KV<String, Long> element, OutputReceiver<KV<String, Long>> receiver) {
+ if (filter.matcher(element.getKey()).matches()) {
// Log at the "DEBUG" level each element that we match. When executing this pipeline
// these log lines will appear only if the log level is set to "DEBUG" or lower.
- LOG.debug("Matched: {}", c.element().getKey());
+ LOG.debug("Matched: {}", element.getKey());
matchedWords.inc();
- c.output(c.element());
+ receiver.output(element);
} else {
// Log at the "TRACE" level each element that is not matched. Different log levels
// can be used to control the verbosity of logging providing an effective mechanism
// to filter less important information.
- LOG.trace("Did not match: {}", c.element().getKey());
+ LOG.trace("Did not match: {}", element.getKey());
unmatchedWords.inc();
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/DistinctExample.java b/examples/java/src/main/java/org/apache/beam/examples/DistinctExample.java
index d3ff9d663f4..3dc5c776f2b 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/DistinctExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/DistinctExample.java
@@ -23,6 +23,8 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.Distinct;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.slf4j.Logger;
@@ -33,7 +35,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Distinct transform usage.
// multifile: false
// default_example: false
-// context_line: 46
+// context_line: 48
// categories:
// - Core Transforms
// complexity: BASIC
@@ -68,9 +70,9 @@ public class DistinctExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/FlatMapElementsExample.java b/examples/java/src/main/java/org/apache/beam/examples/FlatMapElementsExample.java
index 71d05ac7ade..d944ab75ebf 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/FlatMapElementsExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/FlatMapElementsExample.java
@@ -24,6 +24,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.FlatMapElements;
import org.apache.beam.sdk.transforms.InferableFunction;
import org.apache.beam.sdk.transforms.ParDo;
@@ -36,7 +38,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of FlatMapElements transform usage.
// multifile: false
// default_example: false
-// context_line: 50
+// context_line: 52
// categories:
// - Core Transforms
// complexity: BASIC
@@ -78,9 +80,9 @@ public class FlatMapElementsExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/GroupIntoBatchesExample.java b/examples/java/src/main/java/org/apache/beam/examples/GroupIntoBatchesExample.java
index 78e898b9c17..012b951fe8a 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/GroupIntoBatchesExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/GroupIntoBatchesExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.GroupIntoBatches;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of GroupIntoBatches transform usage.
// multifile: false
// default_example: false
-// context_line: 47
+// context_line: 49
// categories:
// - Core Transforms
// complexity: BASIC
@@ -74,9 +76,9 @@ public class GroupIntoBatchesExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/KafkaPassengerCountJson.java b/examples/java/src/main/java/org/apache/beam/examples/KafkaPassengerCountJson.java
index 20f70232ae9..144093b704a 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/KafkaPassengerCountJson.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/KafkaPassengerCountJson.java
@@ -22,7 +22,7 @@ package org.apache.beam.examples;
// description: Read NYC Taxi dataset from Kafka server to count passengers for each vendor
// multifile: false
// default_example: false
-// context_line: 64
+// context_line: 81
// categories:
// - Emulated Data Source
// - IO
@@ -55,6 +55,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.transforms.Values;
@@ -109,10 +111,13 @@ public class KafkaPassengerCountJson {
ParDo.of(
new DoFn<String, KV<Integer, Integer>>() {
@ProcessElement
- public void processElement(ProcessContext c) throws JsonProcessingException {
+ public void processElement(
+ @Element String element, OutputReceiver<KV<Integer, Integer>> receiver)
+ throws JsonProcessingException {
final VendorToPassengerDTO result =
- om.readValue(c.element(), new TypeReference<VendorToPassengerDTO>() {});
- c.output(KV.of(result.getVendorIdField(), result.getPassengerCountField()));
+ om.readValue(element, new TypeReference<VendorToPassengerDTO>() {});
+ receiver.output(
+ KV.of(result.getVendorIdField(), result.getPassengerCountField()));
}
}))
.apply(
@@ -124,11 +129,11 @@ public class KafkaPassengerCountJson {
new DoFn<KV<Integer, Integer>, KV<Integer, Integer>>() {
@ProcessElement
public void processElement(
- ProcessContext c, OutputReceiver<KV<Integer, Integer>> out) {
+ OutputReceiver<KV<Integer, Integer>> out,
+ @Element KV<Integer, Integer> element) {
System.out.printf(
- "Vendor: %s, Passengers: %s%n",
- c.element().getKey(), c.element().getValue());
- out.output(c.element());
+ "Vendor: %s, Passengers: %s%n", element.getKey(), element.getValue());
+ out.output(element);
}
}));
p.run().waitUntilFinish();
diff --git a/examples/java/src/main/java/org/apache/beam/examples/KafkaStreaming.java b/examples/java/src/main/java/org/apache/beam/examples/KafkaStreaming.java
index f0b09226865..83c36d18925 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/KafkaStreaming.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/KafkaStreaming.java
@@ -21,7 +21,7 @@ package org.apache.beam.examples;
// name: KafkaStreaming
// description: Example of streaming data processing using Kafka
// multifile: false
-// context_line: 186
+// context_line: 113
// never_run: true
// always_run: true
// categories:
@@ -49,6 +49,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Sum;
@@ -175,8 +177,9 @@ public class KafkaStreaming {
private static final Random RANDOM = new Random();
@ProcessElement
- public void processElement(ProcessContext c) {
- c.output(generate());
+ public void processElement(
+ @Element Object element, OutputReceiver<KV<String, Integer>> receiver) {
+ receiver.output(generate());
}
public KV<String, Integer> generate() {
@@ -290,17 +293,17 @@ public class KafkaStreaming {
static class LogResults extends DoFn<Map<String, Integer>, Map<String, Integer>> {
@ProcessElement
- public void processElement(ProcessContext c, IntervalWindow w) throws Exception {
- Map<String, Integer> map = c.element();
- if (map == null) {
- c.output(c.element());
- return;
- }
+ public void processElement(
+ PaneInfo pane,
+ IntervalWindow w,
+ @Element Map<String, Integer> scores,
+ OutputReceiver<Map<String, Integer>> receiver)
+ throws Exception {
String startTime = w.start().toString(dateTimeFormatter);
String endTime = w.end().toString(dateTimeFormatter);
- PaneInfo.Timing timing = c.pane().getTiming();
+ PaneInfo.Timing timing = pane.getTiming();
switch (timing) {
case EARLY:
@@ -316,7 +319,7 @@ public class KafkaStreaming {
throw new RuntimeException("Unknown timing value");
}
- for (Map.Entry<String, Integer> entry : map.entrySet()) {
+ for (Map.Entry<String, Integer> entry : scores.entrySet()) {
System.out.printf("%10s: %-10s%n", entry.getKey(), entry.getValue());
}
@@ -326,7 +329,7 @@ public class KafkaStreaming {
System.out.println();
}
- c.output(c.element());
+ receiver.output(scores);
}
}
@@ -340,9 +343,9 @@ public class KafkaStreaming {
static class LogErrorFn extends DoFn<BadRecord, BadRecord> {
@ProcessElement
- public void processElement(@Element BadRecord record, OutputReceiver<BadRecord> receiver) {
- System.out.println(record);
- receiver.output(record);
+ public void processElement(@Element BadRecord badRecord, OutputReceiver<BadRecord> receiver) {
+ System.out.println(badRecord);
+ receiver.output(badRecord);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/KafkaWordCountAvro.java b/examples/java/src/main/java/org/apache/beam/examples/KafkaWordCountAvro.java
index 9e2da248017..50a5e746d0c 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/KafkaWordCountAvro.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/KafkaWordCountAvro.java
@@ -21,7 +21,7 @@ package org.apache.beam.examples;
// name: KafkaWordCountAvro
// description: Read CountWords dataset (CountWords.avro) from Kafka to count words
// multifile: false
-// context_line: 64
+// context_line: 66
// categories:
// - Emulated Data Source
// - IO
@@ -51,6 +51,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SimpleFunction;
@@ -103,10 +105,11 @@ public class KafkaWordCountAvro {
ParDo.of(
new DoFn<String, String>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- for (String word : c.element().split(TOKENIZER_PATTERN, 0)) {
+ public void processElement(
+ @Element String element, OutputReceiver<String> receiver) {
+ for (String word : element.split(TOKENIZER_PATTERN, 0)) {
if (!word.isEmpty()) {
- c.output(word);
+ receiver.output(word);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/KafkaWordCountJson.java b/examples/java/src/main/java/org/apache/beam/examples/KafkaWordCountJson.java
index 355614b4386..e0d0976a105 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/KafkaWordCountJson.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/KafkaWordCountJson.java
@@ -21,7 +21,7 @@ package org.apache.beam.examples;
// name: KafkaWordCountJson
// description: Read CountWords dataset (CountWords.json) from Kafka to count words
// multifile: false
-// context_line: 65
+// context_line: 67
// categories:
// - Emulated Data Source
// - IO
@@ -52,6 +52,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SimpleFunction;
@@ -104,10 +106,11 @@ public class KafkaWordCountJson {
ParDo.of(
new DoFn<String, String>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- for (String word : c.element().split(TOKENIZER_PATTERN, 0)) {
+ public void processElement(
+ @Element String element, OutputReceiver<String> receiver) {
+ for (String word : element.split(TOKENIZER_PATTERN, 0)) {
if (!word.isEmpty()) {
- c.output(word);
+ receiver.output(word);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/KeysExample.java b/examples/java/src/main/java/org/apache/beam/examples/KeysExample.java
index 155834bc0a4..725fb3586fe 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/KeysExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/KeysExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Keys;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Keys transform usage.
// multifile: false
// default_example: false
-// context_line: 46
+// context_line: 48
// categories:
// - Core Transforms
// complexity: BASIC
@@ -70,9 +72,9 @@ public class KeysExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/KvSwapExample.java b/examples/java/src/main/java/org/apache/beam/examples/KvSwapExample.java
index 090779de7c3..f91623f24d3 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/KvSwapExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/KvSwapExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.KvSwap;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of KvSwap transform usage.
// multifile: false
// default_example: false
-// context_line: 46
+// context_line: 48
// categories:
// - Core Transforms
// complexity: BASIC
@@ -69,9 +71,9 @@ public class KvSwapExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/LatestExample.java b/examples/java/src/main/java/org/apache/beam/examples/LatestExample.java
index 5e9662a2f1a..691727382ce 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/LatestExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/LatestExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Latest;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.WithTimestamps;
@@ -36,7 +38,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Latest transform usage.
// multifile: false
// default_example: false
-// context_line: 49
+// context_line: 51
// categories:
// - Core Transforms
// complexity: BASIC
@@ -84,9 +86,9 @@ public class LatestExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/MapElementsExample.java b/examples/java/src/main/java/org/apache/beam/examples/MapElementsExample.java
index 93d885750ae..08a11ccbbb1 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/MapElementsExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/MapElementsExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SimpleFunction;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of MapElements transform usage.
// multifile: false
// default_example: false
-// context_line: 47
+// context_line: 49
// categories:
// - Core Transforms
// complexity: BASIC
@@ -76,9 +78,9 @@ public class MapElementsExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/MaxExample.java b/examples/java/src/main/java/org/apache/beam/examples/MaxExample.java
index 9173d11754a..ba164aa6c43 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/MaxExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/MaxExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Max;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
@@ -33,7 +35,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Max transform usage.
// multifile: false
// default_example: false
-// context_line: 45
+// context_line: 47
// categories:
// - Core Transforms
// complexity: BASIC
@@ -63,9 +65,9 @@ public class MaxExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/MaxPerKeyExample.java b/examples/java/src/main/java/org/apache/beam/examples/MaxPerKeyExample.java
index f5eda817992..ae0a60664ed 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/MaxPerKeyExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/MaxPerKeyExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Max;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Max.perKey transform usage.
// multifile: false
// default_example: false
-// context_line: 47
+// context_line: 49
// categories:
// - Core Transforms
// complexity: BASIC
@@ -67,9 +69,9 @@ public class MaxPerKeyExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/MeanExample.java b/examples/java/src/main/java/org/apache/beam/examples/MeanExample.java
index a31907977df..506d4a6196f 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/MeanExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/MeanExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Mean;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
@@ -33,7 +35,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Mean transform usage.
// multifile: false
// default_example: false
-// context_line: 45
+// context_line: 47
// categories:
// - Core Transforms
// complexity: BASIC
@@ -63,9 +65,9 @@ public class MeanExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/MeanPerKeyExample.java b/examples/java/src/main/java/org/apache/beam/examples/MeanPerKeyExample.java
index aecccee067e..a1e3a1f6ff5 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/MeanPerKeyExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/MeanPerKeyExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Mean;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Mean.perKey transform usage.
// multifile: false
// default_example: false
-// context_line: 47
+// context_line: 49
// categories:
// - Core Transforms
// complexity: BASIC
@@ -67,9 +69,9 @@ public class MeanPerKeyExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/MinExample.java b/examples/java/src/main/java/org/apache/beam/examples/MinExample.java
index a76bcdc5ee3..17a4e0d6554 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/MinExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/MinExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Min;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
@@ -33,7 +35,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Min transform usage.
// multifile: false
// default_example: false
-// context_line: 45
+// context_line: 47
// categories:
// - Core Transforms
// complexity: BASIC
@@ -63,9 +65,9 @@ public class MinExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/MinPerKeyExample.java b/examples/java/src/main/java/org/apache/beam/examples/MinPerKeyExample.java
index d3c0312feaf..6fb8a0468ae 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/MinPerKeyExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/MinPerKeyExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Min;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Min.perKey transform usage.
// multifile: false
// default_example: false
-// context_line: 47
+// context_line: 49
// categories:
// - Core Transforms
// complexity: BASIC
@@ -67,9 +69,9 @@ public class MinPerKeyExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("LogOutput: {} {}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("LogOutput: {} {}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/PartitionExample.java b/examples/java/src/main/java/org/apache/beam/examples/PartitionExample.java
index b34f2bdd16b..780c795b087 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/PartitionExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/PartitionExample.java
@@ -32,6 +32,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Partition;
import org.apache.beam.sdk.values.PCollection;
@@ -44,7 +46,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Partition transform usage.
// multifile: false
// default_example: false
-// context_line: 58
+// context_line: 60
// categories:
// - Core Transforms
// - Coders
@@ -192,9 +194,9 @@ public class PartitionExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/RateLimitedPubSubReader.java b/examples/java/src/main/java/org/apache/beam/examples/RateLimitedPubSubReader.java
index b1156dabee3..060a9195e6a 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/RateLimitedPubSubReader.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/RateLimitedPubSubReader.java
@@ -104,8 +104,8 @@ public class RateLimitedPubSubReader {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- String element = c.element();
+ public void processElement(@Element String element, OutputReceiver<String> receiver)
+ throws Exception {
try {
Preconditions.checkNotNull(rateLimiter).allow(1);
} catch (Exception e) {
@@ -115,7 +115,7 @@ public class RateLimitedPubSubReader {
// Simulate external API call or simply log the read entry
Thread.sleep(100);
LOG.info("Received and rate-limited record: {}", element);
- c.output(element);
+ receiver.output(element);
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/RateLimiterSimple.java b/examples/java/src/main/java/org/apache/beam/examples/RateLimiterSimple.java
index 3ec8fcec0bd..57b7a084849 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/RateLimiterSimple.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/RateLimiterSimple.java
@@ -31,6 +31,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -98,8 +100,8 @@ public class RateLimiterSimple {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- String element = c.element();
+ public void processElement(@Element String element, OutputReceiver<String> receiver)
+ throws Exception {
try {
Preconditions.checkNotNull(rateLimiter).allow(1);
} catch (Exception e) {
@@ -109,7 +111,7 @@ public class RateLimiterSimple {
// Simulate external API call
LOG.info("Processing: {}", element);
Thread.sleep(100);
- c.output("Processed: " + element);
+ receiver.output("Processed: " + element);
}
}
// [END RateLimiterSimpleJava]
diff --git a/examples/java/src/main/java/org/apache/beam/examples/RegexExample.java b/examples/java/src/main/java/org/apache/beam/examples/RegexExample.java
index a0d467718be..c71ce8bb9b4 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/RegexExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/RegexExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Regex;
import org.apache.beam.sdk.values.PCollection;
@@ -33,7 +35,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Regex transform usage.
// multifile: false
// default_example: false
-// context_line: 46
+// context_line: 48
// categories:
// - Core Transforms
// complexity: BASIC
@@ -75,9 +77,9 @@ public class RegexExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/SampleExample.java b/examples/java/src/main/java/org/apache/beam/examples/SampleExample.java
index ed1d90606d3..d5db80403d9 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/SampleExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/SampleExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Sample;
import org.apache.beam.sdk.values.KV;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Sample transform usage.
// multifile: false
// default_example: false
-// context_line: 47
+// context_line: 49
// categories:
// - Core Transforms
// complexity: BASIC
@@ -76,9 +78,9 @@ public class SampleExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/SumExample.java b/examples/java/src/main/java/org/apache/beam/examples/SumExample.java
index 00fcc869792..58c7cc9d2d2 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/SumExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/SumExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.values.PCollection;
@@ -33,7 +35,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Sum transform usage.
// multifile: false
// default_example: false
-// context_line: 45
+// context_line: 47
// categories:
// - Core Transforms
// complexity: BASIC
@@ -63,9 +65,9 @@ public class SumExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/SumPerKeyExample.java b/examples/java/src/main/java/org/apache/beam/examples/SumPerKeyExample.java
index 45d4a9ffd85..8cc4baa2b4a 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/SumPerKeyExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/SumPerKeyExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.values.KV;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Sum.integersPerKey transform usage.
// multifile: false
// default_example: false
-// context_line: 47
+// context_line: 49
// categories:
// - Core Transforms
// complexity: BASIC
@@ -67,9 +69,9 @@ public class SumPerKeyExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/ToStringExample.java b/examples/java/src/main/java/org/apache/beam/examples/ToStringExample.java
index 23e3db6cfd9..b2020ee65ab 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/ToStringExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/ToStringExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.ToString;
import org.apache.beam.sdk.values.KV;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of ToString transform usage.
// multifile: false
// default_example: false
-// context_line: 46
+// context_line: 48
// categories:
// - Core Transforms
// complexity: BASIC
@@ -74,9 +76,9 @@ public class ToStringExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/TopExample.java b/examples/java/src/main/java/org/apache/beam/examples/TopExample.java
index 520af0f6655..fa76f1308e3 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/TopExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/TopExample.java
@@ -23,6 +23,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Top;
import org.apache.beam.sdk.values.PCollection;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Top transform usage.
// multifile: false
// default_example: false
-// context_line: 46
+// context_line: 48
// categories:
// - Core Transforms
// complexity: BASIC
@@ -64,9 +66,9 @@ public class TopExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/ValuesExample.java b/examples/java/src/main/java/org/apache/beam/examples/ValuesExample.java
index 3fc9e84fcb3..9502bd6824d 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/ValuesExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/ValuesExample.java
@@ -22,6 +22,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Values;
import org.apache.beam.sdk.values.KV;
@@ -34,7 +36,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Values transform usage.
// multifile: false
// default_example: false
-// context_line: 46
+// context_line: 48
// categories:
// - Core Transforms
// complexity: BASIC
@@ -70,9 +72,9 @@ public class ValuesExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/ViewExample.java b/examples/java/src/main/java/org/apache/beam/examples/ViewExample.java
index 01c1b1c3f31..39e15affcfa 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/ViewExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/ViewExample.java
@@ -23,6 +23,9 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.SideInput;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.View;
import org.apache.beam.sdk.values.KV;
@@ -36,7 +39,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of View transform usage.
// multifile: false
// default_example: false
-// context_line: 49
+// context_line: 52
// categories:
// - Core Transforms
// complexity: MEDIUM
@@ -83,10 +86,8 @@ public class ViewExample {
@ProcessElement
public void processElement(
@Element KV<String, String> person,
- OutputReceiver<KV<String, String>> out,
- ProcessContext context) {
- Map<String, String> citiesToCountries =
- context.sideInput(citiesToCountriesView);
+ @SideInput("citiesToCountries") Map<String, String> citiesToCountries,
+ OutputReceiver<KV<String, String>> out) {
String city = person.getValue();
String country = citiesToCountries.get(city);
if (country == null) {
@@ -95,7 +96,7 @@ public class ViewExample {
out.output(KV.of(person.getKey(), country));
}
})
- .withSideInputs(citiesToCountriesView));
+ .withSideInput("citiesToCountries", citiesToCountriesView));
// [END main_section]
output.apply("Log", ParDo.of(new LogOutput<>("Output: ")));
@@ -112,9 +113,9 @@ public class ViewExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/WindowExample.java b/examples/java/src/main/java/org/apache/beam/examples/WindowExample.java
index 244a42bc961..a669a0eca77 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/WindowExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/WindowExample.java
@@ -25,6 +25,8 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
@@ -39,7 +41,7 @@ import org.slf4j.LoggerFactory;
// description: Demonstration of Window transform usage.
// multifile: false
// default_example: false
-// context_line: 54
+// context_line: 56
// categories:
// - Core Transforms
// - Windowing
@@ -87,9 +89,9 @@ public class WindowExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java b/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
index 11114043e83..24a3c1b16dc 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
@@ -52,6 +52,8 @@ import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Filter;
import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.PTransform;
@@ -131,10 +133,11 @@ public class AutoComplete {
ParDo.of(
new DoFn<KV<String, Long>, CompletionCandidate>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- c.output(
- new CompletionCandidate(
- c.element().getKey(), c.element().getValue()));
+ public void processElement(
+ @Element KV<String, Long> element,
+ OutputReceiver<CompletionCandidate> receiver) {
+ receiver.output(
+ new CompletionCandidate(element.getKey(), element.getValue()));
}
}));
@@ -210,9 +213,11 @@ public class AutoComplete {
private static class FlattenTops
extends DoFn<KV<String, List<CompletionCandidate>>, CompletionCandidate> {
@ProcessElement
- public void processElement(ProcessContext c) {
- for (CompletionCandidate cc : c.element().getValue()) {
- c.output(cc);
+ public void processElement(
+ @Element KV<String, List<CompletionCandidate>> element,
+ OutputReceiver<CompletionCandidate> receiver) {
+ for (CompletionCandidate cc : element.getValue()) {
+ receiver.output(cc);
}
}
}
@@ -267,10 +272,12 @@ public class AutoComplete {
}
@ProcessElement
- public void processElement(ProcessContext c) {
- String word = c.element().value;
+ public void processElement(
+ @Element CompletionCandidate element,
+ OutputReceiver<KV<String, CompletionCandidate>> receiver) {
+ String word = element.value;
for (int i = minPrefix; i <= Math.min(word.length(), maxPrefix); i++) {
- c.output(KV.of(word.substring(0, i), c.element()));
+ receiver.output(KV.of(word.substring(0, i), element));
}
}
}
@@ -332,23 +339,24 @@ public class AutoComplete {
/** Takes as input a set of strings, and emits each #hashtag found therein. */
static class ExtractHashtags extends DoFn<String, String> {
@ProcessElement
- public void processElement(ProcessContext c) {
- Matcher m = Pattern.compile("#\\S+").matcher(c.element());
+ public void processElement(@Element String element, OutputReceiver<String> receiver) {
+ Matcher m = Pattern.compile("#\\S+").matcher(element);
while (m.find()) {
- c.output(m.group().substring(1));
+ receiver.output(m.group().substring(1));
}
}
}
static class FormatForBigquery extends DoFn<KV<String, List<CompletionCandidate>>, TableRow> {
@ProcessElement
- public void processElement(ProcessContext c) {
+ public void processElement(
+ @Element KV<String, List<CompletionCandidate>> element, OutputReceiver<TableRow> receiver) {
List<TableRow> completions = new ArrayList<>();
- for (CompletionCandidate cc : c.element().getValue()) {
+ for (CompletionCandidate cc : element.getValue()) {
completions.add(new TableRow().set("count", cc.getCount()).set("tag", cc.getValue()));
}
- TableRow row = new TableRow().set("prefix", c.element().getKey()).set("tags", completions);
- c.output(row);
+ TableRow row = new TableRow().set("prefix", element.getKey()).set("tags", completions);
+ receiver.output(row);
}
/** Defines the BigQuery schema used for the output. */
@@ -386,15 +394,16 @@ public class AutoComplete {
}
@ProcessElement
- public void processElement(ProcessContext c) {
+ public void processElement(
+ @Element KV<String, List<CompletionCandidate>> element, OutputReceiver<Entity> receiver) {
Entity.Builder entityBuilder = Entity.newBuilder();
com.google.datastore.v1.Key key =
- makeKey(makeKey(kind, ancestorKey).build(), kind, c.element().getKey()).build();
+ makeKey(makeKey(kind, ancestorKey).build(), kind, element.getKey()).build();
entityBuilder.setKey(key);
List<Value> candidates = new ArrayList<>();
Map<String, Value> properties = new HashMap<>();
- for (CompletionCandidate tag : c.element().getValue()) {
+ for (CompletionCandidate tag : element.getValue()) {
Entity.Builder tagEntity = Entity.newBuilder();
properties.put("tag", makeValue(tag.value).build());
properties.put("count", makeValue(tag.count).build());
@@ -402,7 +411,7 @@ public class AutoComplete {
}
properties.put("candidates", makeValue(candidates).build());
entityBuilder.putAllProperties(properties);
- c.output(entityBuilder.build());
+ receiver.output(entityBuilder.build());
}
}
@@ -527,11 +536,12 @@ public class AutoComplete {
ParDo.of(
new DoFn<KV<String, List<CompletionCandidate>>, Long>() {
@ProcessElement
- public void process(ProcessContext c) {
- KV<String, List<CompletionCandidate>> elm = c.element();
+ public void process(
+ @Element KV<String, List<CompletionCandidate>> elm,
+ OutputReceiver<Long> receiver) {
Long listHash =
- c.element().getValue().stream().mapToLong(cc -> cc.hashCode()).sum();
- c.output(Long.valueOf(elm.getKey().hashCode()) + listHash);
+ elm.getValue().stream().mapToLong(cc -> cc.hashCode()).sum();
+ receiver.output(Long.valueOf(elm.getKey().hashCode()) + listHash);
}
}))
.apply(Sum.longsGlobally());
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java b/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
index e4ce5e3eb17..ea5f2e5399e 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
@@ -34,6 +34,8 @@ import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
/**
@@ -55,12 +57,12 @@ public class StreamingWordExtract {
/** A {@link DoFn} that tokenizes lines of text into individual words. */
static class ExtractWords extends DoFn<String, String> {
@ProcessElement
- public void processElement(ProcessContext c) {
- String[] words = c.element().split(ExampleUtils.TOKENIZER_PATTERN, -1);
+ public void processElement(@Element String element, OutputReceiver<String> receiver) {
+ String[] words = element.split(ExampleUtils.TOKENIZER_PATTERN, -1);
for (String word : words) {
if (!word.isEmpty()) {
- c.output(word);
+ receiver.output(word);
}
}
}
@@ -69,8 +71,8 @@ public class StreamingWordExtract {
/** A {@link DoFn} that uppercases a word. */
static class Uppercase extends DoFn<String, String> {
@ProcessElement
- public void processElement(ProcessContext c) {
- c.output(c.element().toUpperCase());
+ public void processElement(@Element String element, OutputReceiver<String> receiver) {
+ receiver.output(element.toUpperCase());
}
}
@@ -78,8 +80,8 @@ public class StreamingWordExtract {
static class StringToRowConverter extends DoFn<String, TableRow> {
/** In this example, put the whole string into single BigQuery field. */
@ProcessElement
- public void processElement(ProcessContext c) {
- c.output(new TableRow().set("string_field", c.element()));
+ public void processElement(@Element String element, OutputReceiver<TableRow> receiver) {
+ receiver.output(new TableRow().set("string_field", element));
}
static TableSchema getSchema() {
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/TfIdf.java b/examples/java/src/main/java/org/apache/beam/examples/complete/TfIdf.java
index b3e5fd04fa7..2b5c8b42ee1 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/TfIdf.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/TfIdf.java
@@ -23,7 +23,7 @@ package org.apache.beam.examples.complete;
// GCS prefix.
// multifile: true
// pipeline_options: --output output.txt
-// context_line: 447
+// context_line: 462
// categories:
// - Combiners
// - Options
@@ -56,6 +56,9 @@ import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Distinct;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.SideInput;
import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.Keys;
import org.apache.beam.sdk.transforms.PTransform;
@@ -236,9 +239,11 @@ public class TfIdf {
ParDo.of(
new DoFn<KV<URI, String>, KV<URI, String>>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- URI uri = c.element().getKey();
- String line = c.element().getValue();
+ public void processElement(
+ @Element KV<URI, String> element,
+ OutputReceiver<KV<URI, String>> receiver) {
+ URI uri = element.getKey();
+ String line = element.getValue();
for (String word : line.split("\\W+", -1)) {
// Log INFO messages when the word “love” is found.
if ("love".equalsIgnoreCase(word)) {
@@ -246,7 +251,7 @@ public class TfIdf {
}
if (!word.isEmpty()) {
- c.output(KV.of(uri, word.toLowerCase()));
+ receiver.output(KV.of(uri, word.toLowerCase()));
}
}
}
@@ -281,11 +286,13 @@ public class TfIdf {
ParDo.of(
new DoFn<KV<KV<URI, String>, Long>, KV<URI, KV<String, Long>>>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- URI uri = c.element().getKey().getKey();
- String word = c.element().getKey().getValue();
- Long occurrences = c.element().getValue();
- c.output(KV.of(uri, KV.of(word, occurrences)));
+ public void processElement(
+ @Element KV<KV<URI, String>, Long> element,
+ OutputReceiver<KV<URI, KV<String, Long>>> receiver) {
+ URI uri = element.getKey().getKey();
+ String word = element.getKey().getValue();
+ Long occurrences = element.getValue();
+ receiver.output(KV.of(uri, KV.of(word, occurrences)));
}
}));
@@ -322,16 +329,18 @@ public class TfIdf {
ParDo.of(
new DoFn<KV<URI, CoGbkResult>, KV<String, KV<URI, Double>>>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- URI uri = c.element().getKey();
- Long wordTotal = c.element().getValue().getOnly(wordTotalsTag);
+ public void processElement(
+ @Element KV<URI, CoGbkResult> element,
+ OutputReceiver<KV<String, KV<URI, Double>>> receiver) {
+ URI uri = element.getKey();
+ Long wordTotal = element.getValue().getOnly(wordTotalsTag);
for (KV<String, Long> wordAndCount :
- c.element().getValue().getAll(wordCountsTag)) {
+ element.getValue().getAll(wordCountsTag)) {
String word = wordAndCount.getKey();
Long wordCount = wordAndCount.getValue();
Double termFrequency = wordCount.doubleValue() / wordTotal.doubleValue();
- c.output(KV.of(word, KV.of(uri, termFrequency)));
+ receiver.output(KV.of(word, KV.of(uri, termFrequency)));
}
}
}));
@@ -348,17 +357,19 @@ public class TfIdf {
ParDo.of(
new DoFn<KV<String, Long>, KV<String, Double>>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- String word = c.element().getKey();
- Long documentCount = c.element().getValue();
- Long documentTotal = c.sideInput(totalDocuments);
+ public void processElement(
+ @SideInput("totalDocuments") Long documentTotal,
+ @Element KV<String, Long> element,
+ OutputReceiver<KV<String, Double>> receiver) {
+ String word = element.getKey();
+ Long documentCount = element.getValue();
Double documentFrequency =
documentCount.doubleValue() / documentTotal.doubleValue();
- c.output(KV.of(word, documentFrequency));
+ receiver.output(KV.of(word, documentFrequency));
}
})
- .withSideInputs(totalDocuments));
+ .withSideInput("totalDocuments", totalDocuments));
// Join the term frequency and document frequency
// collections, each keyed on the word.
@@ -380,15 +391,17 @@ public class TfIdf {
ParDo.of(
new DoFn<KV<String, CoGbkResult>, KV<String, KV<URI, Double>>>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- String word = c.element().getKey();
- Double df = c.element().getValue().getOnly(dfTag);
+ public void processElement(
+ @Element KV<String, CoGbkResult> element,
+ OutputReceiver<KV<String, KV<URI, Double>>> receiver) {
+ String word = element.getKey();
+ Double df = element.getValue().getOnly(dfTag);
- for (KV<URI, Double> uriAndTf : c.element().getValue().getAll(tfTag)) {
+ for (KV<URI, Double> uriAndTf : element.getValue().getAll(tfTag)) {
URI uri = uriAndTf.getKey();
Double tf = uriAndTf.getValue();
Double tfIdf = tf * Math.log(1 / df);
- c.output(KV.of(word, KV.of(uri, tfIdf)));
+ receiver.output(KV.of(word, KV.of(uri, tfIdf)));
}
}
}));
@@ -419,13 +432,15 @@ public class TfIdf {
ParDo.of(
new DoFn<KV<String, KV<URI, Double>>, String>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- c.output(
+ public void processElement(
+ @Element KV<String, KV<URI, Double>> element,
+ OutputReceiver<String> receiver) {
+ receiver.output(
String.format(
"%s,\t%s,\t%f",
- c.element().getKey(),
- c.element().getValue().getKey(),
- c.element().getValue().getValue()));
+ element.getKey(),
+ element.getValue().getKey(),
+ element.getValue().getValue()));
}
}))
.apply(TextIO.write().to(output).withSuffix(".csv"));
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/TopWikipediaSessions.java b/examples/java/src/main/java/org/apache/beam/examples/complete/TopWikipediaSessions.java
index b06cd8da9d4..e79f09c9d08 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/TopWikipediaSessions.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/TopWikipediaSessions.java
@@ -24,7 +24,7 @@ package org.apache.beam.examples.complete;
// each month.
// multifile: true
// pipeline_options: --output output.txt
-// context_line: 236
+// context_line: 244
// categories:
// - Combiners
// - Options
@@ -48,6 +48,8 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -93,8 +95,7 @@ public class TopWikipediaSessions {
/** Extracts user and timestamp from a TableRow representing a Wikipedia edit. */
static class ExtractUserAndTimestamp extends DoFn<TableRow, String> {
@ProcessElement
- public void processElement(ProcessContext c) {
- TableRow row = c.element();
+ public void processElement(@Element TableRow row, OutputReceiver<String> receiver) {
int timestamp;
// TODO(BEAM-5390): Avoid this workaround.
try {
@@ -105,7 +106,7 @@ public class TopWikipediaSessions {
String userName = (String) row.get("contributor_username");
if (userName != null) {
// Sets the implicit timestamp field to be used in windowing.
- c.outputWithTimestamp(userName, new Instant(timestamp * 1000L));
+ receiver.outputWithTimestamp(userName, new Instant(timestamp * 1000L));
}
}
}
@@ -143,18 +144,24 @@ public class TopWikipediaSessions {
static class SessionsToStringsDoFn extends DoFn<KV<String, Long>, KV<String, Long>> {
@ProcessElement
- public void processElement(ProcessContext c, BoundedWindow window) {
- c.output(KV.of(c.element().getKey() + " : " + window, c.element().getValue()));
+ public void processElement(
+ BoundedWindow window,
+ @Element KV<String, Long> element,
+ OutputReceiver<KV<String, Long>> receiver) {
+ receiver.output(KV.of(element.getKey() + " : " + window, element.getValue()));
}
}
static class FormatOutputDoFn extends DoFn<List<KV<String, Long>>, String> {
@ProcessElement
- public void processElement(ProcessContext c, BoundedWindow window) {
- for (KV<String, Long> item : c.element()) {
+ public void processElement(
+ BoundedWindow window,
+ @Element List<KV<String, Long>> element,
+ OutputReceiver<String> receiver) {
+ for (KV<String, Long> item : element) {
String session = item.getKey();
long count = item.getValue();
- c.output(session + " : " + count + " : " + ((IntervalWindow) window).start());
+ receiver.output(session + " : " + count + " : " + ((IntervalWindow) window).start());
}
}
}
@@ -187,10 +194,11 @@ public class TopWikipediaSessions {
ParDo.of(
new DoFn<String, String>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- if (Math.abs((long) c.element().hashCode())
+ public void processElement(
+ @Element String element, OutputReceiver<String> receiver) {
+ if (Math.abs((long) element.hashCode())
<= Integer.MAX_VALUE * samplingThreshold) {
- c.output(c.element());
+ receiver.output(element);
}
}
}))
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficMaxLaneFlow.java b/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficMaxLaneFlow.java
index 6f75e2e03d9..ab80a220f57 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficMaxLaneFlow.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficMaxLaneFlow.java
@@ -23,7 +23,7 @@ package org.apache.beam.examples.complete;
// window, it finds the lane that had the highest flow recorded, for each sensor station.
// It writes those max values along with auxiliary info to a BigQuery table.
// multifile: true
-// context_line: 402
+// context_line: 410
// categories:
// - Combiners
// - Streaming
@@ -58,6 +58,9 @@ import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.Timestamp;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SerializableFunction;
@@ -182,13 +185,14 @@ public class TrafficMaxLaneFlow {
DateTimeFormat.forPattern("MM/dd/yyyy HH:mm:ss");
@ProcessElement
- public void processElement(DoFn<String, String>.ProcessContext c) throws Exception {
- String[] items = c.element().split(",", -1);
+ public void processElement(@Element String element, OutputReceiver<String> receiver)
+ throws Exception {
+ String[] items = element.split(",", -1);
if (items.length > 0) {
try {
String timestamp = items[0];
- c.outputWithTimestamp(c.element(), new Instant(dateTimeFormat.parseMillis(timestamp)));
+ receiver.outputWithTimestamp(element, new Instant(dateTimeFormat.parseMillis(timestamp)));
} catch (IllegalArgumentException e) {
// Skip the invalid input.
}
@@ -206,8 +210,9 @@ public class TrafficMaxLaneFlow {
static class ExtractFlowInfoFn extends DoFn<String, KV<String, LaneInfo>> {
@ProcessElement
- public void processElement(ProcessContext c) {
- String[] items = c.element().split(",", -1);
+ public void processElement(
+ @Element String element, OutputReceiver<KV<String, LaneInfo>> receiver) {
+ String[] items = element.split(",", -1);
if (items.length < 48) {
// Skip the invalid input.
return;
@@ -236,7 +241,7 @@ public class TrafficMaxLaneFlow {
laneAvgOccupancy,
laneAvgSpeed,
totalFlow);
- c.output(KV.of(stationId, laneInfo));
+ receiver.output(KV.of(stationId, laneInfo));
}
}
}
@@ -270,12 +275,15 @@ public class TrafficMaxLaneFlow {
*/
static class FormatMaxesFn extends DoFn<KV<String, LaneInfo>, TableRow> {
@ProcessElement
- public void processElement(ProcessContext c) {
+ public void processElement(
+ @Element KV<String, LaneInfo> element,
+ @Timestamp Instant timestamp,
+ OutputReceiver<TableRow> receiver) {
- LaneInfo laneInfo = c.element().getValue();
+ LaneInfo laneInfo = element.getValue();
TableRow row =
new TableRow()
- .set("station_id", c.element().getKey())
+ .set("station_id", element.getKey())
.set("direction", laneInfo.getDirection())
.set("freeway", laneInfo.getFreeway())
.set("lane_max_flow", laneInfo.getLaneFlow())
@@ -284,8 +292,8 @@ public class TrafficMaxLaneFlow {
.set("avg_speed", laneInfo.getLaneAS())
.set("total_flow", laneInfo.getTotalFlow())
.set("recorded_timestamp", laneInfo.getRecordedTimestamp())
- .set("window_timestamp", c.timestamp().toString());
- c.output(row);
+ .set("window_timestamp", timestamp.toString());
+ receiver.output(row);
}
/** Defines the BigQuery schema used for the output. */
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficRoutes.java b/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficRoutes.java
index 95841562686..3918f2d72a6 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficRoutes.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficRoutes.java
@@ -24,7 +24,7 @@ package org.apache.beam.examples.complete;
// of predefined 'routes', and looks for 'slowdowns' in those routes. It writes its
// results to a BigQuery table.
// multifile: true
-// context_line: 399
+// context_line: 411
// categories:
// - Combiners
// - Streaming
@@ -63,6 +63,9 @@ import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.Timestamp;
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -187,12 +190,13 @@ public class TrafficRoutes {
DateTimeFormat.forPattern("MM/dd/yyyy HH:mm:ss");
@ProcessElement
- public void processElement(DoFn<String, String>.ProcessContext c) throws Exception {
- String[] items = c.element().split(",");
+ public void processElement(@Element String element, OutputReceiver<String> receiver)
+ throws Exception {
+ String[] items = element.split(",");
String timestamp = tryParseTimestamp(items);
if (timestamp != null) {
try {
- c.outputWithTimestamp(c.element(), new Instant(dateTimeFormat.parseMillis(timestamp)));
+ receiver.outputWithTimestamp(element, new Instant(dateTimeFormat.parseMillis(timestamp)));
} catch (IllegalArgumentException e) {
// Skip the invalid input.
}
@@ -207,8 +211,11 @@ public class TrafficRoutes {
static class ExtractStationSpeedFn extends DoFn<String, KV<String, StationSpeed>> {
@ProcessElement
- public void processElement(ProcessContext c) {
- String[] items = c.element().split(",");
+ public void processElement(
+ @Timestamp Instant timestamp,
+ @Element String element,
+ OutputReceiver<KV<String, StationSpeed>> receiver) {
+ String[] items = element.split(",");
String stationType = tryParseStationType(items);
// For this analysis, use only 'main line' station types
if ("ML".equals(stationType)) {
@@ -216,11 +223,10 @@ public class TrafficRoutes {
String stationId = tryParseStationId(items);
// For this simple example, filter out everything but some hardwired routes.
if (avgSpeed != null && stationId != null && sdStations.containsKey(stationId)) {
- StationSpeed stationSpeed =
- new StationSpeed(stationId, avgSpeed, c.timestamp().getMillis());
+ StationSpeed stationSpeed = new StationSpeed(stationId, avgSpeed, timestamp.getMillis());
// The tuple key is the 'route' name stored in the 'sdStations' hash.
KV<String, StationSpeed> outputValue = KV.of(sdStations.get(stationId), stationSpeed);
- c.output(outputValue);
+ receiver.output(outputValue);
}
}
}
@@ -234,13 +240,16 @@ public class TrafficRoutes {
*/
static class GatherStats extends DoFn<KV<String, Iterable<StationSpeed>>, KV<String, RouteInfo>> {
@ProcessElement
- public void processElement(ProcessContext c) throws IOException {
- String route = c.element().getKey();
+ public void processElement(
+ @Element KV<String, Iterable<StationSpeed>> element,
+ OutputReceiver<KV<String, RouteInfo>> receiver)
+ throws IOException {
+ String route = element.getKey();
double speedSum = 0.0;
int speedCount = 0;
int speedups = 0;
int slowdowns = 0;
- List<StationSpeed> infoList = Lists.newArrayList(c.element().getValue());
+ List<StationSpeed> infoList = Lists.newArrayList(element.getValue());
// StationSpeeds sort by embedded timestamp.
Collections.sort(infoList);
Map<String, Double> prevSpeeds = new HashMap<>();
@@ -268,22 +277,25 @@ public class TrafficRoutes {
double speedAvg = speedSum / speedCount;
boolean slowdownEvent = slowdowns >= 2 * speedups;
RouteInfo routeInfo = new RouteInfo(route, speedAvg, slowdownEvent);
- c.output(KV.of(route, routeInfo));
+ receiver.output(KV.of(route, routeInfo));
}
}
/** Format the results of the slowdown calculations to a TableRow, to save to BigQuery. */
static class FormatStatsFn extends DoFn<KV<String, RouteInfo>, TableRow> {
@ProcessElement
- public void processElement(ProcessContext c) {
- RouteInfo routeInfo = c.element().getValue();
+ public void processElement(
+ @Element KV<String, RouteInfo> element,
+ @Timestamp Instant timestamp,
+ OutputReceiver<TableRow> receiver) {
+ RouteInfo routeInfo = element.getValue();
TableRow row =
new TableRow()
.set("avg_speed", routeInfo.getAvgSpeed())
.set("slowdown_event", routeInfo.getSlowdownEvent())
- .set("route", c.element().getKey())
- .set("window_timestamp", c.timestamp().toString());
- c.output(row);
+ .set("route", element.getKey())
+ .set("window_timestamp", timestamp.toString());
+ receiver.output(row);
}
/** Defines the BigQuery schema used for the output. */
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/DataProtectors.java b/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/DataProtectors.java
index cf097c8ea97..6e5321dc54a 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/DataProtectors.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/DataProtectors.java
@@ -37,6 +37,9 @@ import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.Schema.Field;
import org.apache.beam.sdk.schemas.Schema.FieldType;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.MultiOutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.GroupIntoBatches;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -199,20 +202,24 @@ public class DataProtectors {
@ProcessElement
@SuppressWarnings("argument")
- public void process(@Element KV<Integer, Iterable<Row>> element, ProcessContext context) {
+ public void process(
+ @Element KV<Integer, Iterable<Row>> element,
+ OutputReceiver<Row> mainReceiver,
+ MultiOutputReceiver multiReceiver) {
Iterable<Row> rows = element.getValue();
try {
for (Row outputRow : getTokenizedRow(rows)) {
- context.output(outputRow);
+ mainReceiver.output(outputRow);
}
} catch (Exception e) {
for (Row outputRow : rows) {
- context.output(
- failureTag,
- FailsafeElement.of(outputRow, outputRow)
- .setErrorMessage(e.getMessage())
- .setStacktrace(Throwables.getStackTraceAsString(e)));
+ multiReceiver
+ .get(failureTag)
+ .output(
+ FailsafeElement.of(outputRow, outputRow)
+ .setErrorMessage(e.getMessage())
+ .setStacktrace(Throwables.getStackTraceAsString(e)));
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/io/TokenizationBigQueryIO.java b/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/io/TokenizationBigQueryIO.java
index fe8f4c1afad..667e26fa9c2 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/io/TokenizationBigQueryIO.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/io/TokenizationBigQueryIO.java
@@ -26,6 +26,8 @@ import org.apache.beam.sdk.io.gcp.bigquery.BigQueryUtils;
import org.apache.beam.sdk.io.gcp.bigquery.InsertRetryPolicy;
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.Row;
@@ -90,9 +92,8 @@ public class TokenizationBigQueryIO {
public static class RowToTableRowFn extends DoFn<Row, TableRow> {
@ProcessElement
- public void processElement(ProcessContext context) {
- Row row = context.element();
- context.output(BigQueryUtils.toTableRow(row));
+ public void processElement(@Element Row row, OutputReceiver<TableRow> receiver) {
+ receiver.output(BigQueryUtils.toTableRow(row));
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/io/TokenizationBigTableIO.java b/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/io/TokenizationBigTableIO.java
index d7d1c3e9723..435620a4b01 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/io/TokenizationBigTableIO.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/transforms/io/TokenizationBigTableIO.java
@@ -75,8 +75,10 @@ public class TokenizationBigTableIO {
@ProcessElement
public void processElement(
- @Element Row in, OutputReceiver<KV<ByteString, Iterable<Mutation>>> out, ProcessContext c) {
- DataTokenizationOptions options = c.getPipelineOptions().as(DataTokenizationOptions.class);
+ @Element Row in,
+ OutputReceiver<KV<ByteString, Iterable<Mutation>>> out,
+ PipelineOptions pipelineOptions) {
+ DataTokenizationOptions options = pipelineOptions.as(DataTokenizationOptions.class);
// Mapping every field in provided Row to Mutation.SetCell, which will create/update
// cell content with provided data
Set<Mutation> mutations =
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/utils/CsvConverters.java b/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/utils/CsvConverters.java
index d827e4b30cb..9fead4e1a7f 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/utils/CsvConverters.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/utils/CsvConverters.java
@@ -264,8 +264,6 @@ public class CsvConverters {
@Override
public PCollectionTuple expand(PCollectionTuple lines) {
- PCollectionView<String> headersView = null;
-
// Convert csv lines into Failsafe elements so that we can recover over multiple transforms.
PCollection<FailsafeElement<String, String>> lineFailsafeElements =
lines
@@ -285,16 +283,14 @@ public class CsvConverters {
return lineFailsafeElements.apply(
"LineToDocumentUsingSchema",
- ParDo.of(
- new FailsafeElementToJsonFn(
- headersView, schema, delimiter(), udfDeadletterTag()))
+ ParDo.of(new FailsafeElementToJsonFn(schema, delimiter(), udfDeadletterTag()))
.withOutputTags(udfOutputTag(), TupleTagList.of(udfDeadletterTag())));
}
// Run if using headers
- headersView = lines.get(headerTag()).apply(Sample.any(1)).apply(View.asSingleton());
+ PCollectionView<String> headersView =
+ lines.get(headerTag()).apply(Sample.any(1)).apply(View.asSingleton());
- PCollectionView<String> finalHeadersView = headersView;
lines
.get(headerTag())
.apply(
@@ -302,23 +298,24 @@ public class CsvConverters {
ParDo.of(
new DoFn<String, String>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- String headers = c.sideInput(finalHeadersView);
- if (!c.element().equals(headers)) {
+ public void processElement(
+ @SideInput("finalHeadersView") String headers,
+ @Element String element) {
+ if (!element.equals(headers)) {
LOG.error("Headers do not match, consistency cannot be guaranteed");
throw new RuntimeException(
"Headers do not match, consistency cannot be guaranteed");
}
}
})
- .withSideInputs(finalHeadersView));
+ .withSideInput("finalHeadersView", headersView));
return lineFailsafeElements.apply(
"LineToDocumentWithHeaders",
ParDo.of(
- new FailsafeElementToJsonFn(
- headersView, jsonSchemaPath(), delimiter(), udfDeadletterTag()))
- .withSideInputs(headersView)
+ new FailsafeElementToJsonWithHeadersFn(
+ jsonSchemaPath(), delimiter(), udfDeadletterTag()))
+ .withSideInput("finalHeadersView", headersView)
.withOutputTags(udfOutputTag(), TupleTagList.of(udfDeadletterTag())));
}
@@ -357,45 +354,90 @@ public class CsvConverters {
@Nullable public final String jsonSchema;
public final String delimiter;
public final TupleTag<FailsafeElement<String, String>> udfDeadletterTag;
- @Nullable private final PCollectionView<String> headersView;
private Counter successCounter =
Metrics.counter(FailsafeElementToJsonFn.class, SUCCESSFUL_TO_JSON_COUNTER);
private Counter failedCounter =
Metrics.counter(FailsafeElementToJsonFn.class, FAILED_TO_JSON_COUNTER);
FailsafeElementToJsonFn(
- PCollectionView<String> headersView,
String jsonSchema,
String delimiter,
TupleTag<FailsafeElement<String, String>> udfDeadletterTag) {
- this.headersView = headersView;
this.jsonSchema = jsonSchema;
this.delimiter = delimiter;
this.udfDeadletterTag = udfDeadletterTag;
}
@ProcessElement
- public void processElement(ProcessContext context) {
- FailsafeElement<String, String> element = context.element();
+ public void processElement(
+ @Element FailsafeElement<String, String> element,
+ OutputReceiver<FailsafeElement<String, String>> receiver,
+ MultiOutputReceiver multiReceiver) {
+ List<String> header = null;
+ List<String> record = Arrays.asList(element.getOriginalPayload().split(this.delimiter));
+
+ try {
+ String json = buildJsonString(header, record, this.jsonSchema);
+ receiver.output(FailsafeElement.of(element.getOriginalPayload(), json));
+ successCounter.inc();
+ } catch (Exception e) {
+ failedCounter.inc();
+ multiReceiver
+ .get(this.udfDeadletterTag)
+ .output(
+ FailsafeElement.of(element)
+ .setErrorMessage(e.getMessage())
+ .setStacktrace(Throwables.getStackTraceAsString(e)));
+ }
+ }
+ }
+
+ public static class FailsafeElementToJsonWithHeadersFn
+ extends DoFn<FailsafeElement<String, String>, FailsafeElement<String, String>> {
+
+ @Nullable public final String jsonSchema;
+ public final String delimiter;
+ public final TupleTag<FailsafeElement<String, String>> udfDeadletterTag;
+ private Counter successCounter =
+ Metrics.counter(FailsafeElementToJsonWithHeadersFn.class, SUCCESSFUL_TO_JSON_COUNTER);
+ private Counter failedCounter =
+ Metrics.counter(FailsafeElementToJsonWithHeadersFn.class, FAILED_TO_JSON_COUNTER);
+
+ FailsafeElementToJsonWithHeadersFn(
+ String jsonSchema,
+ String delimiter,
+ TupleTag<FailsafeElement<String, String>> udfDeadletterTag) {
+ this.jsonSchema = jsonSchema;
+ this.delimiter = delimiter;
+ this.udfDeadletterTag = udfDeadletterTag;
+ }
+
+ @ProcessElement
+ public void processElement(
+ @Element FailsafeElement<String, String> element,
+ OutputReceiver<FailsafeElement<String, String>> receiver,
+ MultiOutputReceiver multiReceiver,
+ @SideInput("finalHeadersView") String headersStr) {
List<String> header = null;
- if (this.headersView != null) {
- header = Arrays.asList(context.sideInput(this.headersView).split(this.delimiter));
+ if (headersStr != null) {
+ header = Arrays.asList(headersStr.split(this.delimiter));
}
List<String> record = Arrays.asList(element.getOriginalPayload().split(this.delimiter));
try {
String json = buildJsonString(header, record, this.jsonSchema);
- context.output(FailsafeElement.of(element.getOriginalPayload(), json));
+ receiver.output(FailsafeElement.of(element.getOriginalPayload(), json));
successCounter.inc();
} catch (Exception e) {
failedCounter.inc();
- context.output(
- this.udfDeadletterTag,
- FailsafeElement.of(element)
- .setErrorMessage(e.getMessage())
- .setStacktrace(Throwables.getStackTraceAsString(e)));
+ multiReceiver
+ .get(this.udfDeadletterTag)
+ .output(
+ FailsafeElement.of(element)
+ .setErrorMessage(e.getMessage())
+ .setStacktrace(Throwables.getStackTraceAsString(e)));
}
}
}
@@ -407,9 +449,9 @@ public class CsvConverters {
static class LineToFailsafeElementFn extends DoFn<String, FailsafeElement<String, String>> {
@ProcessElement
- public void processElement(ProcessContext context) {
- String message = context.element();
- context.output(FailsafeElement.of(message, message));
+ public void processElement(
+ @Element String message, OutputReceiver<FailsafeElement<String, String>> receiver) {
+ receiver.output(FailsafeElement.of(message, message));
}
}
@@ -510,13 +552,12 @@ public class CsvConverters {
}
@ProcessElement
- public void processElement(ProcessContext context, MultiOutputReceiver outputReceiver) {
- ReadableFile f = context.element();
+ public void processElement(@Element ReadableFile file, MultiOutputReceiver outputReceiver) {
String headers;
List<String> records = null;
String delimiter = String.valueOf(this.csvFormat.getDelimiter());
try {
- String csvFileString = f.readFullyAsUTF8String();
+ String csvFileString = file.readFullyAsUTF8String();
StringReader reader = new StringReader(csvFileString);
CSVParser parser = CSVParser.parse(reader, this.csvFormat.withFirstRecordAsHeader());
records =
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/utils/ErrorConverters.java b/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/utils/ErrorConverters.java
index 01377add085..99981cb17b8 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/utils/ErrorConverters.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/datatokenization/utils/ErrorConverters.java
@@ -29,6 +29,9 @@ import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.Timestamp;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -43,6 +46,7 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjec
import org.checkerframework.checker.nullness.qual.Nullable;
import org.joda.time.DateTimeZone;
import org.joda.time.Duration;
+import org.joda.time.Instant;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
@@ -123,16 +127,17 @@ public class ErrorConverters {
}
@ProcessElement
- public void processElement(ProcessContext context) {
- FailsafeElement<String, String> failsafeElement = context.element();
+ public void processElement(
+ @Element FailsafeElement<String, String> failsafeElement,
+ @Timestamp Instant timestamp,
+ OutputReceiver<String> receiver) {
ArrayList<String> outputRow = new ArrayList<>();
final String message = failsafeElement.getOriginalPayload();
// Format the timestamp for insertion
- String timestamp =
- TIMESTAMP_FORMATTER.print(context.timestamp().toDateTime(DateTimeZone.UTC));
+ String timestampStr = TIMESTAMP_FORMATTER.print(timestamp.toDateTime(DateTimeZone.UTC));
- outputRow.add(timestamp);
+ outputRow.add(timestampStr);
outputRow.add(MoreObjects.firstNonNull(failsafeElement.getErrorMessage(), ""));
// Only set the payload if it's populated on the message.
@@ -140,7 +145,7 @@ public class ErrorConverters {
outputRow.add(message);
}
- context.output(String.join(csvDelimiter, outputRow));
+ receiver.output(String.join(csvDelimiter, outputRow));
}
}
@@ -198,19 +203,20 @@ public class ErrorConverters {
DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");
@ProcessElement
- public void processElement(ProcessContext context) {
- FailsafeElement<String, String> failsafeElement = context.element();
+ public void processElement(
+ @Timestamp Instant timestamp,
+ @Element FailsafeElement<String, String> failsafeElement,
+ OutputReceiver<TableRow> receiver) {
final String message = failsafeElement.getOriginalPayload();
// Format the timestamp for insertion
- String timestamp =
- TIMESTAMP_FORMATTER.print(context.timestamp().toDateTime(DateTimeZone.UTC));
+ String timestampStr = TIMESTAMP_FORMATTER.print(timestamp.toDateTime(DateTimeZone.UTC));
// Build the table row
@SuppressWarnings("nullness") // TableRow.set not annotated but does accept nulls
final TableRow failedRow =
new TableRow()
- .set("timestamp", timestamp)
+ .set("timestamp", timestampStr)
.set("errorMessage", failsafeElement.getErrorMessage())
.set("stacktrace", failsafeElement.getStacktrace());
@@ -221,7 +227,7 @@ public class ErrorConverters {
.set("payloadBytes", message.getBytes(StandardCharsets.UTF_8));
}
- context.output(failedRow);
+ receiver.output(failedRow);
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/game/GameStats.java b/examples/java/src/main/java/org/apache/beam/examples/complete/game/GameStats.java
index a3ed04bb1c4..0018201f368 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/game/GameStats.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/game/GameStats.java
@@ -34,6 +34,9 @@ import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.SideInput;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.Mean;
import org.apache.beam.sdk.transforms.PTransform;
@@ -126,21 +129,23 @@ public class GameStats extends LeaderBoard {
Metrics.counter("main", "SpammerUsers");
@ProcessElement
- public void processElement(ProcessContext c) {
- Integer score = c.element().getValue();
- Double gmc = c.sideInput(globalMeanScore);
+ public void processElement(
+ @SideInput("globalMeanScore") Double gmc,
+ @Element KV<String, Integer> element,
+ OutputReceiver<KV<String, Integer>> receiver) {
+ Integer score = element.getValue();
if (score > (gmc * SCORE_WEIGHT)) {
LOG.info(
"user {} spammer score {} with mean {}",
- c.element().getKey(),
+ element.getKey(),
score,
gmc);
numSpammerUsers.inc();
- c.output(c.element());
+ receiver.output(element);
}
}
})
- .withSideInputs(globalMeanScore));
+ .withSideInput("globalMeanScore", globalMeanScore));
return filtered;
}
}
@@ -149,10 +154,10 @@ public class GameStats extends LeaderBoard {
/** Calculate and output an element's session duration. */
private static class UserSessionInfoFn extends DoFn<KV<String, Integer>, Integer> {
@ProcessElement
- public void processElement(ProcessContext c, BoundedWindow window) {
+ public void processElement(BoundedWindow window, OutputReceiver<Integer> receiver) {
IntervalWindow w = (IntervalWindow) window;
int duration = new Duration(w.start(), w.end()).toPeriod().toStandardMinutes().getMinutes();
- c.output(duration);
+ receiver.output(duration);
}
}
@@ -192,22 +197,21 @@ public class GameStats extends LeaderBoard {
configureWindowedWrite() {
Map<String, WriteToBigQuery.FieldInfo<KV<String, Integer>>> tableConfigure = new HashMap<>();
tableConfigure.put(
- "team", new WriteToBigQuery.FieldInfo<>("STRING", (c, w) -> c.element().getKey()));
+ "team", new WriteToBigQuery.FieldInfo<>("STRING", (e, w, t, p) -> e.getKey()));
tableConfigure.put(
- "total_score",
- new WriteToBigQuery.FieldInfo<>("INTEGER", (c, w) -> c.element().getValue()));
+ "total_score", new WriteToBigQuery.FieldInfo<>("INTEGER", (e, w, t, p) -> e.getValue()));
tableConfigure.put(
"window_start",
new WriteToBigQuery.FieldInfo<>(
"STRING",
- (c, w) -> {
+ (e, w, t, p) -> {
IntervalWindow window = (IntervalWindow) w;
return GameConstants.DATE_TIME_FORMATTER.print(window.start());
}));
tableConfigure.put(
"processing_time",
new WriteToBigQuery.FieldInfo<>(
- "STRING", (c, w) -> GameConstants.DATE_TIME_FORMATTER.print(Instant.now())));
+ "STRING", (e, w, t, p) -> GameConstants.DATE_TIME_FORMATTER.print(Instant.now())));
return tableConfigure;
}
@@ -222,12 +226,12 @@ public class GameStats extends LeaderBoard {
"window_start",
new WriteToBigQuery.FieldInfo<>(
"STRING",
- (c, w) -> {
+ (e, w, t, p) -> {
IntervalWindow window = (IntervalWindow) w;
return GameConstants.DATE_TIME_FORMATTER.print(window.start());
}));
tableConfigure.put(
- "mean_duration", new WriteToBigQuery.FieldInfo<>("FLOAT", (c, w) -> c.element()));
+ "mean_duration", new WriteToBigQuery.FieldInfo<>("FLOAT", (e, w, t, p) -> e));
return tableConfigure;
}
@@ -288,14 +292,17 @@ public class GameStats extends LeaderBoard {
ParDo.of(
new DoFn<GameActionInfo, GameActionInfo>() {
@ProcessElement
- public void processElement(ProcessContext c) {
+ public void processElement(
+ @SideInput("spammersView") Map<String, Integer> spammers,
+ @Element GameActionInfo element,
+ OutputReceiver<GameActionInfo> receiver) {
// If the user is not in the spammers Map, output the data element.
- if (c.sideInput(spammersView).get(c.element().getUser().trim()) == null) {
- c.output(c.element());
+ if (spammers.get(element.getUser().trim()) == null) {
+ receiver.output(element);
}
}
})
- .withSideInputs(spammersView))
+ .withSideInput("spammersView", spammersView))
// Extract and sum teamname/score pairs from the event data.
.apply("ExtractTeamScore", new ExtractAndSumScore("team"))
// [END DocInclude_FilterAndCalc]
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/game/HourlyTeamScore.java b/examples/java/src/main/java/org/apache/beam/examples/complete/game/HourlyTeamScore.java
index c57d9ba6b8c..a94f699282a 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/game/HourlyTeamScore.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/game/HourlyTeamScore.java
@@ -112,11 +112,11 @@ public class HourlyTeamScore extends UserScore {
*/
protected static Map<String, WriteToText.FieldFn<KV<String, Integer>>> configureOutput() {
Map<String, WriteToText.FieldFn<KV<String, Integer>>> config = new HashMap<>();
- config.put("team", (c, w) -> c.element().getKey());
- config.put("total_score", (c, w) -> c.element().getValue());
+ config.put("team", (e, w, t, p) -> e.getKey());
+ config.put("total_score", (e, w, t, p) -> e.getValue());
config.put(
"window_start",
- (c, w) -> {
+ (e, w, t, p) -> {
IntervalWindow window = (IntervalWindow) w;
return GameConstants.DATE_TIME_FORMATTER.print(window.start());
});
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/game/LeaderBoard.java b/examples/java/src/main/java/org/apache/beam/examples/complete/game/LeaderBoard.java
index 832c0ad79e7..3750b33aa79 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/game/LeaderBoard.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/game/LeaderBoard.java
@@ -135,25 +135,24 @@ public class LeaderBoard extends HourlyTeamScore {
Map<String, WriteToBigQuery.FieldInfo<KV<String, Integer>>> tableConfigure = new HashMap<>();
tableConfigure.put(
- "team", new WriteToBigQuery.FieldInfo<>("STRING", (c, w) -> c.element().getKey()));
+ "team", new WriteToBigQuery.FieldInfo<>("STRING", (e, w, t, p) -> e.getKey()));
tableConfigure.put(
- "total_score",
- new WriteToBigQuery.FieldInfo<>("INTEGER", (c, w) -> c.element().getValue()));
+ "total_score", new WriteToBigQuery.FieldInfo<>("INTEGER", (e, w, t, p) -> e.getValue()));
tableConfigure.put(
"window_start",
new WriteToBigQuery.FieldInfo<>(
"STRING",
- (c, w) -> {
+ (e, w, t, p) -> {
IntervalWindow window = (IntervalWindow) w;
return GameConstants.DATE_TIME_FORMATTER.print(window.start());
}));
tableConfigure.put(
"processing_time",
new WriteToBigQuery.FieldInfo<>(
- "STRING", (c, w) -> GameConstants.DATE_TIME_FORMATTER.print(Instant.now())));
+ "STRING", (e, w, t, p) -> GameConstants.DATE_TIME_FORMATTER.print(Instant.now())));
tableConfigure.put(
"timing",
- new WriteToBigQuery.FieldInfo<>("STRING", (c, w) -> c.pane().getTiming().toString()));
+ new WriteToBigQuery.FieldInfo<>("STRING", (e, w, t, p) -> p.getTiming().toString()));
return tableConfigure;
}
@@ -165,10 +164,9 @@ public class LeaderBoard extends HourlyTeamScore {
configureBigQueryWrite() {
Map<String, WriteToBigQuery.FieldInfo<KV<String, Integer>>> tableConfigure = new HashMap<>();
tableConfigure.put(
- "user", new WriteToBigQuery.FieldInfo<>("STRING", (c, w) -> c.element().getKey()));
+ "user", new WriteToBigQuery.FieldInfo<>("STRING", (e, w, t, p) -> e.getKey()));
tableConfigure.put(
- "total_score",
- new WriteToBigQuery.FieldInfo<>("INTEGER", (c, w) -> c.element().getValue()));
+ "total_score", new WriteToBigQuery.FieldInfo<>("INTEGER", (e, w, t, p) -> e.getValue()));
return tableConfigure;
}
@@ -184,7 +182,7 @@ public class LeaderBoard extends HourlyTeamScore {
tableConfigure.put(
"processing_time",
new WriteToBigQuery.FieldInfo<>(
- "STRING", (c, w) -> GameConstants.DATE_TIME_FORMATTER.print(Instant.now())));
+ "STRING", (e, w, t, p) -> GameConstants.DATE_TIME_FORMATTER.print(Instant.now())));
return tableConfigure;
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/game/StatefulTeamScore.java b/examples/java/src/main/java/org/apache/beam/examples/complete/game/StatefulTeamScore.java
index b28db261ab0..3caa1e61952 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/game/StatefulTeamScore.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/game/StatefulTeamScore.java
@@ -37,6 +37,8 @@ import org.apache.beam.sdk.state.StateSpec;
import org.apache.beam.sdk.state.StateSpecs;
import org.apache.beam.sdk.state.ValueState;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
@@ -101,12 +103,12 @@ public class StatefulTeamScore extends LeaderBoard {
private static Map<String, FieldInfo<KV<String, Integer>>> configureCompleteWindowedTableWrite() {
Map<String, FieldInfo<KV<String, Integer>>> tableConfigure = new HashMap<>();
- tableConfigure.put("team", new FieldInfo<>("STRING", (c, w) -> c.element().getKey()));
- tableConfigure.put("total_score", new FieldInfo<>("INTEGER", (c, w) -> c.element().getValue()));
+ tableConfigure.put("team", new FieldInfo<>("STRING", (e, w, t, p) -> e.getKey()));
+ tableConfigure.put("total_score", new FieldInfo<>("INTEGER", (e, w, t, p) -> e.getValue()));
tableConfigure.put(
"processing_time",
new FieldInfo<>(
- "STRING", (c, w) -> GameConstants.DATE_TIME_FORMATTER.print(Instant.now())));
+ "STRING", (e, w, t, p) -> GameConstants.DATE_TIME_FORMATTER.print(Instant.now())));
return tableConfigure;
}
@@ -203,9 +205,11 @@ public class StatefulTeamScore extends LeaderBoard {
*/
@ProcessElement
public void processElement(
- ProcessContext c, @StateId(TOTAL_SCORE) ValueState<Integer> totalScore) {
- String teamName = c.element().getKey();
- GameActionInfo gInfo = c.element().getValue();
+ @Element KV<String, GameActionInfo> element,
+ @StateId(TOTAL_SCORE) ValueState<Integer> totalScore,
+ OutputReceiver<KV<String, Integer>> receiver) {
+ String teamName = element.getKey();
+ GameActionInfo gInfo = element.getValue();
// ValueState cells do not contain a default value. If the state is possibly not written, make
// sure to check for null on read.
@@ -218,7 +222,7 @@ public class StatefulTeamScore extends LeaderBoard {
// the new total is 2002, and the threshold is 1000, 1999 / 1000 = 1, 2002 / 1000 = 2.
// Therefore, this team passed the threshold.
if (oldTotalScore / this.thresholdScore < totalScore.read() / this.thresholdScore) {
- c.output(KV.of(teamName, totalScore.read()));
+ receiver.output(KV.of(teamName, totalScore.read()));
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/game/UserScore.java b/examples/java/src/main/java/org/apache/beam/examples/complete/game/UserScore.java
index b30b4665d26..054ce7a5293 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/game/UserScore.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/game/UserScore.java
@@ -34,6 +34,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -161,18 +163,18 @@ public class UserScore {
private final Counter numParseErrors = Metrics.counter("main", "ParseErrors");
@ProcessElement
- public void processElement(ProcessContext c) {
- String[] components = c.element().split(",", -1);
+ public void processElement(@Element String element, OutputReceiver<GameActionInfo> receiver) {
+ String[] components = element.split(",", -1);
try {
String user = components[0].trim();
String team = components[1].trim();
Integer score = Integer.parseInt(components[2].trim());
Long timestamp = Long.parseLong(components[3].trim());
GameActionInfo gInfo = new GameActionInfo(user, team, score, timestamp);
- c.output(gInfo);
+ receiver.output(gInfo);
} catch (ArrayIndexOutOfBoundsException | NumberFormatException e) {
numParseErrors.inc();
- LOG.info("Parse error on {}", c.element(), e);
+ LOG.info("Parse error on {}", element, e);
}
}
}
@@ -232,8 +234,8 @@ public class UserScore {
*/
protected static Map<String, WriteToText.FieldFn<KV<String, Integer>>> configureOutput() {
Map<String, WriteToText.FieldFn<KV<String, Integer>>> config = new HashMap<>();
- config.put("user", (c, w) -> c.element().getKey());
- config.put("total_score", (c, w) -> c.element().getValue());
+ config.put("user", (e, w, t, p) -> e.getKey());
+ config.put("total_score", (e, w, t, p) -> e.getValue());
return config;
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToBigQuery.java b/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToBigQuery.java
index eef4bc93268..5486025083c 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToBigQuery.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToBigQuery.java
@@ -30,11 +30,16 @@ import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.InsertRetryPolicy;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.Timestamp;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PDone;
+import org.joda.time.Instant;
/**
* Generate, format, and write BigQuery table row information. Use provided information about the
@@ -64,11 +69,11 @@ public class WriteToBigQuery<InputT> extends PTransform<PCollection<InputT>, PDo
}
/**
- * A {@link Serializable} function from a {@link DoFn.ProcessContext} and {@link BoundedWindow} to
- * the value for that field.
+ * A {@link Serializable} function from an element and {@link BoundedWindow} to the value for that
+ * field.
*/
public interface FieldFn<InputT> extends Serializable {
- Object apply(DoFn<InputT, TableRow>.ProcessContext context, BoundedWindow window);
+ Object apply(InputT element, BoundedWindow window, Instant timestamp, PaneInfo pane);
}
/** Define a class to hold information about output table field definitions. */
@@ -96,16 +101,21 @@ public class WriteToBigQuery<InputT> extends PTransform<PCollection<InputT>, PDo
protected class BuildRowFn extends DoFn<InputT, TableRow> {
@ProcessElement
- public void processElement(ProcessContext c, BoundedWindow window) {
+ public void processElement(
+ @Element InputT element,
+ @Timestamp Instant timestamp,
+ PaneInfo pane,
+ BoundedWindow window,
+ OutputReceiver<TableRow> receiver) {
TableRow row = new TableRow();
for (Map.Entry<String, FieldInfo<InputT>> entry : fieldInfo.entrySet()) {
String key = entry.getKey();
FieldInfo<InputT> fcnInfo = entry.getValue();
FieldFn<InputT> fcn = fcnInfo.getFieldFn();
- row.set(key, fcn.apply(c, window));
+ row.set(key, fcn.apply(element, window, timestamp, pane));
}
- c.output(row);
+ receiver.output(row);
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToText.java b/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToText.java
index 330769e0c79..35471163c03 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToText.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToText.java
@@ -32,6 +32,9 @@ import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.Timestamp;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
@@ -70,26 +73,32 @@ public class WriteToText<InputT> extends PTransform<PCollection<InputT>, PDone>
}
/**
- * A {@link Serializable} function from a {@link DoFn.ProcessContext} and {@link BoundedWindow} to
- * the value for that field.
+ * A {@link Serializable} function from an element and {@link BoundedWindow} to the value for that
+ * field.
*/
public interface FieldFn<InputT> extends Serializable {
- Object apply(DoFn<InputT, String>.ProcessContext context, BoundedWindow window);
+ Object apply(
+ InputT element, BoundedWindow window, org.joda.time.Instant timestamp, PaneInfo pane);
}
/** Convert each key/score pair into a row as specified by fieldFn. */
protected class BuildRowFn extends DoFn<InputT, String> {
@ProcessElement
- public void processElement(ProcessContext c, BoundedWindow window) {
+ public void processElement(
+ @Element InputT element,
+ @Timestamp org.joda.time.Instant timestamp,
+ PaneInfo pane,
+ BoundedWindow window,
+ OutputReceiver<String> receiver) {
List<String> fields = new ArrayList<>();
for (Map.Entry<String, FieldFn<InputT>> entry : fieldFn.entrySet()) {
String key = entry.getKey();
FieldFn<InputT> fcn = entry.getValue();
- fields.add(key + ": " + fcn.apply(c, window));
+ fields.add(key + ": " + fcn.apply(element, window, timestamp, pane));
}
String result = fields.stream().collect(Collectors.joining(", "));
- c.output(result);
+ receiver.output(result);
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteWindowedToBigQuery.java b/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteWindowedToBigQuery.java
index 36fa18a34e0..77a59f2c3b6 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteWindowedToBigQuery.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteWindowedToBigQuery.java
@@ -24,8 +24,12 @@ import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.InsertRetryPolicy;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.Timestamp;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PDone;
@@ -44,15 +48,20 @@ public class WriteWindowedToBigQuery<T> extends WriteToBigQuery<T> {
/** Convert each key/score pair into a BigQuery TableRow. */
protected class BuildRowFn extends DoFn<T, TableRow> {
@ProcessElement
- public void processElement(ProcessContext c, BoundedWindow window) {
+ public void processElement(
+ @Element T element,
+ @Timestamp org.joda.time.Instant timestamp,
+ PaneInfo pane,
+ BoundedWindow window,
+ OutputReceiver<TableRow> receiver) {
TableRow row = new TableRow();
for (Map.Entry<String, FieldInfo<T>> entry : fieldInfo.entrySet()) {
String key = entry.getKey();
FieldInfo<T> fcnInfo = entry.getValue();
- row.set(key, fcnInfo.getFieldFn().apply(c, window));
+ row.set(key, fcnInfo.getFieldFn().apply(element, window, timestamp, pane));
}
- c.output(row);
+ receiver.output(row);
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryStreamingTornadoes.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryStreamingTornadoes.java
index 3b2653b7601..d662cde1f00 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryStreamingTornadoes.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryStreamingTornadoes.java
@@ -32,6 +32,9 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.Timestamp;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -85,10 +88,9 @@ public class BigQueryStreamingTornadoes {
*/
static class ExtractTornadoesFn extends DoFn<TableRow, Integer> {
@ProcessElement
- public void processElement(ProcessContext c) {
- TableRow row = c.element();
+ public void processElement(@Element TableRow row, OutputReceiver<Integer> receiver) {
if (Boolean.TRUE.equals(row.get("tornado"))) {
- c.output(Integer.parseInt((String) row.get("month")));
+ receiver.output(Integer.parseInt((String) row.get("month")));
}
}
}
@@ -99,13 +101,16 @@ public class BigQueryStreamingTornadoes {
*/
static class FormatCountsFn extends DoFn<KV<Integer, Long>, TableRow> {
@ProcessElement
- public void processElement(ProcessContext c) {
+ public void processElement(
+ @Element KV<Integer, Long> element,
+ @Timestamp Instant timestamp,
+ OutputReceiver<TableRow> receiver) {
TableRow row =
new TableRow()
- .set("ts", c.timestamp().toString())
- .set("month", c.element().getKey())
- .set("tornado_count", c.element().getValue());
- c.output(row);
+ .set("ts", timestamp.toString())
+ .set("month", element.getKey())
+ .set("tornado_count", element.getValue());
+ receiver.output(row);
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java
index 43d720e3526..67e0661a1a4 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java
@@ -32,6 +32,8 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
@@ -79,10 +81,9 @@ public class BigQueryTornadoes {
*/
static class ExtractTornadoesFn extends DoFn<TableRow, Integer> {
@ProcessElement
- public void processElement(ProcessContext c) {
- TableRow row = c.element();
+ public void processElement(@Element TableRow row, OutputReceiver<Integer> receiver) {
if ((Boolean) row.get("tornado")) {
- c.output(Integer.parseInt((String) row.get("month")));
+ receiver.output(Integer.parseInt((String) row.get("month")));
}
}
}
@@ -93,12 +94,11 @@ public class BigQueryTornadoes {
*/
static class FormatCountsFn extends DoFn<KV<Integer, Long>, TableRow> {
@ProcessElement
- public void processElement(ProcessContext c) {
+ public void processElement(
+ @Element KV<Integer, Long> element, OutputReceiver<TableRow> receiver) {
TableRow row =
- new TableRow()
- .set("month", c.element().getKey())
- .set("tornado_count", c.element().getValue());
- c.output(row);
+ new TableRow().set("month", element.getKey()).set("tornado_count", element.getValue());
+ receiver.output(row);
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java
index 2a581d769dc..5f5b5d06b1f 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java
@@ -33,6 +33,8 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SerializableFunction;
@@ -78,12 +80,11 @@ public class CombinePerKeyExamples {
private final Counter smallerWords = Metrics.counter(ExtractLargeWordsFn.class, "smallerWords");
@ProcessElement
- public void processElement(ProcessContext c) {
- TableRow row = c.element();
+ public void processElement(@Element TableRow row, OutputReceiver<KV<String, String>> receiver) {
String playName = (String) row.get("corpus");
String word = (String) row.get("word");
if (word.length() >= MIN_WORD_LENGTH) {
- c.output(KV.of(word, playName));
+ receiver.output(KV.of(word, playName));
} else {
// Track how many smaller words we're not including. This information will be
// visible in the Monitoring UI.
@@ -98,10 +99,11 @@ public class CombinePerKeyExamples {
*/
static class FormatShakespeareOutputFn extends DoFn<KV<String, String>, TableRow> {
@ProcessElement
- public void processElement(ProcessContext c) {
+ public void processElement(
+ @Element KV<String, String> element, OutputReceiver<TableRow> receiver) {
TableRow row =
- new TableRow().set("word", c.element().getKey()).set("all_plays", c.element().getValue());
- c.output(row);
+ new TableRow().set("word", element.getKey()).set("all_plays", element.getValue());
+ receiver.output(row);
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java
index 9187bb83d7d..26a6659a86b 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java
@@ -31,6 +31,9 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.SideInput;
import org.apache.beam.sdk.transforms.Mean;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -87,8 +90,7 @@ public class FilterExamples {
*/
static class ProjectionFn extends DoFn<TableRow, TableRow> {
@ProcessElement
- public void processElement(ProcessContext c) {
- TableRow row = c.element();
+ public void processElement(@Element TableRow row, OutputReceiver<TableRow> receiver) {
// Grab year, month, day, mean_temp from the row
Integer year = Integer.parseInt((String) row.get("year"));
Integer month = Integer.parseInt((String) row.get("month"));
@@ -101,7 +103,7 @@ public class FilterExamples {
.set("month", month)
.set("day", day)
.set("mean_temp", meanTemp);
- c.output(outRow);
+ receiver.output(outRow);
}
}
@@ -119,12 +121,11 @@ public class FilterExamples {
}
@ProcessElement
- public void processElement(ProcessContext c) {
- TableRow row = c.element();
+ public void processElement(@Element TableRow row, OutputReceiver<TableRow> receiver) {
Integer month;
month = (Integer) row.get("month");
if (month.equals(this.monthFilter)) {
- c.output(row);
+ receiver.output(row);
}
}
}
@@ -135,10 +136,9 @@ public class FilterExamples {
*/
static class ExtractTempFn extends DoFn<TableRow, Double> {
@ProcessElement
- public void processElement(ProcessContext c) {
- TableRow row = c.element();
+ public void processElement(@Element TableRow row, OutputReceiver<Double> receiver) {
Double meanTemp = Double.parseDouble(row.get("mean_temp").toString());
- c.output(meanTemp);
+ receiver.output(meanTemp);
}
}
@@ -178,16 +178,17 @@ public class FilterExamples {
ParDo.of(
new DoFn<TableRow, TableRow>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- Double meanTemp =
- Double.parseDouble(c.element().get("mean_temp").toString());
- Double gTemp = c.sideInput(globalMeanTemp);
+ public void processElement(
+ @SideInput("globalMeanTemp") Double gTemp,
+ @Element TableRow element,
+ OutputReceiver<TableRow> receiver) {
+ Double meanTemp = Double.parseDouble(element.get("mean_temp").toString());
if (meanTemp < gTemp) {
- c.output(c.element());
+ receiver.output(element);
}
}
})
- .withSideInputs(globalMeanTemp));
+ .withSideInput("globalMeanTemp", globalMeanTemp));
return filteredRows;
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/JoinExamples.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/JoinExamples.java
index f78df0c0946..e6f8573705a 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/JoinExamples.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/JoinExamples.java
@@ -26,6 +26,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.join.CoGbkResult;
import org.apache.beam.sdk.transforms.join.CoGroupByKey;
@@ -90,13 +92,14 @@ public class JoinExamples {
ParDo.of(
new DoFn<KV<String, CoGbkResult>, KV<String, String>>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- KV<String, CoGbkResult> e = c.element();
- String countryCode = e.getKey();
- String countryName = e.getValue().getOnly(countryInfoTag);
- for (String eventInfo : c.element().getValue().getAll(eventInfoTag)) {
+ public void processElement(
+ @Element KV<String, CoGbkResult> element,
+ OutputReceiver<KV<String, String>> receiver) {
+ String countryCode = element.getKey();
+ String countryName = element.getValue().getOnly(countryInfoTag);
+ for (String eventInfo : element.getValue().getAll(eventInfoTag)) {
// Generate a string that combines information from both collection values
- c.output(
+ receiver.output(
KV.of(
countryCode,
"Country name: " + countryName + ", Event info: " + eventInfo));
@@ -111,10 +114,11 @@ public class JoinExamples {
ParDo.of(
new DoFn<KV<String, String>, String>() {
@ProcessElement
- public void processElement(ProcessContext c) {
+ public void processElement(
+ @Element KV<String, String> element, OutputReceiver<String> receiver) {
String outputstring =
- "Country code: " + c.element().getKey() + ", " + c.element().getValue();
- c.output(outputstring);
+ "Country code: " + element.getKey() + ", " + element.getValue();
+ receiver.output(outputstring);
}
}));
return formattedResults;
@@ -126,14 +130,13 @@ public class JoinExamples {
*/
static class ExtractEventDataFn extends DoFn<TableRow, KV<String, String>> {
@ProcessElement
- public void processElement(ProcessContext c) {
- TableRow row = c.element();
+ public void processElement(@Element TableRow row, OutputReceiver<KV<String, String>> receiver) {
String countryCode = (String) row.get("ActionGeo_CountryCode");
String sqlDate = (String) row.get("SQLDATE");
String actor1Name = (String) row.get("Actor1Name");
String sourceUrl = (String) row.get("SOURCEURL");
String eventInfo = "Date: " + sqlDate + ", Actor1: " + actor1Name + ", url: " + sourceUrl;
- c.output(KV.of(countryCode, eventInfo));
+ receiver.output(KV.of(countryCode, eventInfo));
}
}
@@ -143,11 +146,10 @@ public class JoinExamples {
*/
static class ExtractCountryInfoFn extends DoFn<TableRow, KV<String, String>> {
@ProcessElement
- public void processElement(ProcessContext c) {
- TableRow row = c.element();
+ public void processElement(@Element TableRow row, OutputReceiver<KV<String, String>> receiver) {
String countryCode = (String) row.get("FIPSCC");
String countryName = (String) row.get("HumanName");
- c.output(KV.of(countryCode, countryName));
+ receiver.output(KV.of(countryCode, countryName));
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java
index 8760d562d04..85df56a5825 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java
@@ -30,6 +30,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.Max;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -73,23 +75,22 @@ public class MaxPerKeyExamples {
*/
static class ExtractTempFn extends DoFn<TableRow, KV<Integer, Double>> {
@ProcessElement
- public void processElement(ProcessContext c) {
- TableRow row = c.element();
+ public void processElement(
+ @Element TableRow row, OutputReceiver<KV<Integer, Double>> receiver) {
Integer month = Integer.parseInt((String) row.get("month"));
Double meanTemp = Double.parseDouble(row.get("mean_temp").toString());
- c.output(KV.of(month, meanTemp));
+ receiver.output(KV.of(month, meanTemp));
}
}
/** Format the results to a TableRow, to save to BigQuery. */
static class FormatMaxesFn extends DoFn<KV<Integer, Double>, TableRow> {
@ProcessElement
- public void processElement(ProcessContext c) {
+ public void processElement(
+ @Element KV<Integer, Double> element, OutputReceiver<TableRow> receiver) {
TableRow row =
- new TableRow()
- .set("month", c.element().getKey())
- .set("max_mean_temp", c.element().getValue());
- c.output(row);
+ new TableRow().set("month", element.getKey()).set("max_mean_temp", element.getValue());
+ receiver.output(row);
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/MinimalBigQueryTornadoes.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/MinimalBigQueryTornadoes.java
index 713af1d5095..62f7b492c72 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/MinimalBigQueryTornadoes.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/MinimalBigQueryTornadoes.java
@@ -26,6 +26,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
@@ -41,7 +43,7 @@ import org.slf4j.LoggerFactory;
// always_run: true
// default_example: false
// pipeline_options: --project apache-beam-testing
-// context_line: 102
+// context_line: 125
// categories:
// - Filtering
// - IO
@@ -73,10 +75,9 @@ public class MinimalBigQueryTornadoes {
*/
static class ExtractTornadoesFn extends DoFn<TableRow, Integer> {
@ProcessElement
- public void processElement(ProcessContext c) {
- TableRow row = c.element();
+ public void processElement(@Element TableRow row, OutputReceiver<Integer> receiver) {
if ((Boolean) row.get("tornado")) {
- c.output(Integer.parseInt((String) row.get("month")));
+ receiver.output(Integer.parseInt((String) row.get("month")));
}
}
}
@@ -87,8 +88,9 @@ public class MinimalBigQueryTornadoes {
*/
static class FormatCountsFn extends DoFn<KV<Integer, Long>, String> {
@ProcessElement
- public void processElement(ProcessContext c) {
- c.output(c.element().getKey() + ": " + c.element().getValue());
+ public void processElement(
+ @Element KV<Integer, Long> element, OutputReceiver<String> receiver) {
+ receiver.output(element.getKey() + ": " + element.getValue());
}
}
@@ -134,9 +136,9 @@ public class MinimalBigQueryTornadoes {
}
@ProcessElement
- public void processElement(ProcessContext c) {
- LOG.info("{}{}", prefix, c.element());
- c.output(c.element());
+ public void processElement(@Element T element, OutputReceiver<T> receiver) {
+ LOG.info("{}{}", prefix, element);
+ receiver.output(element);
}
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/TriggerExample.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/TriggerExample.java
index cd3c6dd8415..5270077ea03 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/TriggerExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/TriggerExample.java
@@ -38,6 +38,9 @@ import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.Timestamp;
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -46,6 +49,7 @@ import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.transforms.windowing.Repeatedly;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.values.KV;
@@ -364,15 +368,18 @@ public class TriggerExample {
new DoFn<KV<String, Iterable<Integer>>, KV<String, String>>() {
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- Iterable<Integer> flows = c.element().getValue();
+ public void processElement(
+ @Element KV<String, Iterable<Integer>> element,
+ OutputReceiver<KV<String, String>> receiver)
+ throws Exception {
+ Iterable<Integer> flows = element.getValue();
Integer sum = 0;
Long numberOfRecords = 0L;
for (Integer value : flows) {
sum += value;
numberOfRecords++;
}
- c.output(KV.of(c.element().getKey(), sum + "," + numberOfRecords));
+ receiver.output(KV.of(element.getKey(), sum + "," + numberOfRecords));
}
}));
PCollection<TableRow> output = results.apply(ParDo.of(new FormatTotalFlow(triggerType)));
@@ -392,21 +399,27 @@ public class TriggerExample {
}
@ProcessElement
- public void processElement(ProcessContext c, BoundedWindow window) throws Exception {
- String[] values = c.element().getValue().split(",", -1);
+ public void processElement(
+ PaneInfo pane,
+ @Timestamp Instant timestamp,
+ BoundedWindow window,
+ @Element KV<String, String> element,
+ OutputReceiver<TableRow> receiver)
+ throws Exception {
+ String[] values = element.getValue().split(",", -1);
TableRow row =
new TableRow()
.set("trigger_type", triggerType)
- .set("freeway", c.element().getKey())
+ .set("freeway", element.getKey())
.set("total_flow", Integer.parseInt(values[0]))
.set("number_of_records", Long.parseLong(values[1]))
.set("window", window.toString())
- .set("isFirst", c.pane().isFirst())
- .set("isLast", c.pane().isLast())
- .set("timing", c.pane().getTiming().toString())
- .set("event_time", c.timestamp().toString())
+ .set("isFirst", pane.isFirst())
+ .set("isLast", pane.isLast())
+ .set("timing", pane.getTiming().toString())
+ .set("event_time", timestamp.toString())
.set("processing_time", Instant.now().toString());
- c.output(row);
+ receiver.output(row);
}
}
@@ -418,8 +431,9 @@ public class TriggerExample {
private static final int VALID_NUM_FIELDS = 50;
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- String[] laneInfo = c.element().split(",", -1);
+ public void processElement(
+ @Element String element, OutputReceiver<KV<String, Integer>> receiver) throws Exception {
+ String[] laneInfo = element.split(",", -1);
if ("timestamp".equals(laneInfo[0])) {
// Header row
return;
@@ -435,7 +449,7 @@ public class TriggerExample {
if (totalFlow == null || totalFlow <= 0) {
return;
}
- c.output(KV.of(freeway, totalFlow));
+ receiver.output(KV.of(freeway, totalFlow));
}
}
@@ -509,7 +523,8 @@ public class TriggerExample {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
+ public void processElement(@Element String element, OutputReceiver<String> receiver)
+ throws Exception {
Instant timestamp = Instant.now();
if (random.nextDouble() < THRESHOLD) {
int range = MAX_DELAY - MIN_DELAY;
@@ -517,7 +532,7 @@ public class TriggerExample {
long delayInMillis = TimeUnit.MINUTES.toMillis(delayInMinutes);
timestamp = new Instant(timestamp.getMillis() - delayInMillis);
}
- c.outputWithTimestamp(c.element(), timestamp);
+ receiver.outputWithTimestamp(element, timestamp);
}
}
diff --git a/examples/java/src/main/java/org/apache/beam/examples/snippets/Snippets.java b/examples/java/src/main/java/org/apache/beam/examples/snippets/Snippets.java
index 4f24c69f74b..99aada20669 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/snippets/Snippets.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/snippets/Snippets.java
@@ -76,6 +76,10 @@ import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFn.BoundedPerElement;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.SideInput;
+import org.apache.beam.sdk.transforms.DoFn.Timestamp;
import org.apache.beam.sdk.transforms.Latest;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
@@ -540,14 +544,14 @@ public class Snippets {
ParDo.of(
new DoFn<KV<String, CoGbkResult>, String>() {
@ProcessElement
- public void processElement(ProcessContext c) {
- KV<String, CoGbkResult> e = c.element();
+ public void processElement(
+ @Element KV<String, CoGbkResult> e, OutputReceiver<String> receiver) {
String name = e.getKey();
Iterable<String> emailsIter = e.getValue().getAll(emailsTag);
Iterable<String> phonesIter = e.getValue().getAll(phonesTag);
String formattedResult =
Snippets.formatCoGbkResults(name, emailsIter, phonesIter);
- c.output(formattedResult);
+ receiver.output(formattedResult);
}
}));
// [END CoGroupByKeyTuple]
@@ -642,20 +646,23 @@ public class Snippets {
new DoFn<Long, KV<Long, Long>>() {
@ProcessElement
- public void process(ProcessContext c, @Timestamp Instant timestamp) {
- Iterable<Map<String, String>> si = c.sideInput(mapIterable);
+ public void process(
+ @Timestamp Instant timestamp,
+ @Element Long element,
+ @SideInput("mapIterable") Iterable<Map<String, String>> si,
+ OutputReceiver<KV<Long, Long>> receiver) {
// Take an element from the side input iterable (likely length 1)
Map<String, String> keyMap = si.iterator().next();
- c.outputWithTimestamp(KV.of(1L, c.element()), Instant.now());
+ receiver.outputWithTimestamp(KV.of(1L, element), Instant.now());
LOG.info(
"Value is {} with timestamp {}, using key A from side input with time {}.",
- c.element(),
+ element,
timestamp.toString(DateTimeFormat.forPattern("HH:mm:ss")),
keyMap.get("Key_A"));
}
})
- .withSideInputs(mapIterable));
+ .withSideInput("mapIterable", mapIterable));
p.run();
}
@@ -701,9 +708,9 @@ public class Snippets {
// Define the DoFn that logs the ValueProvider value.
@ProcessElement
- public void process(ProcessContext c) {
+ public void process(PipelineOptions options) {
- MyOptions ops = c.getPipelineOptions().as(MyOptions.class);
+ MyOptions ops = options.as(MyOptions.class);
// This example logs the ValueProvider value, but you could store it by
// pushing it to an external database.
@@ -943,11 +950,13 @@ public class Snippets {
ParDo.of(
new DoFn<Instant, Long>() {
@ProcessElement
- public void process(ProcessContext c) {
- c.output((long) c.sideInput(sideInput).size());
+ public void process(
+ @SideInput("sideInput") List<Long> sideInputValue,
+ OutputReceiver<Long> receiver) {
+ receiver.output((long) sideInputValue.size());
}
})
- .withSideInputs(sideInput));
+ .withSideInput("sideInput", sideInput));
// [END PeriodicallyUpdatingSideInputs]
return result;
}
@@ -1188,7 +1197,10 @@ public class Snippets {
private static class BundleFinalizationDoFn extends DoFn<String, Integer> {
// [START BundleFinalize]
@ProcessElement
- public void processElement(ProcessContext c, BundleFinalizer bundleFinalizer) {
+ public void processElement(
+ @Element String element,
+ OutputReceiver<Integer> receiver,
+ BundleFinalizer bundleFinalizer) {
// ... produce output ...
bundleFinalizer.afterBundleCommit(
diff --git a/examples/java/src/main/java/org/apache/beam/examples/subprocess/ExampleEchoPipeline.java b/examples/java/src/main/java/org/apache/beam/examples/subprocess/ExampleEchoPipeline.java
index 4aa20fc10df..b7fb7b82db3 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/subprocess/ExampleEchoPipeline.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/subprocess/ExampleEchoPipeline.java
@@ -28,6 +28,8 @@ import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.slf4j.Logger;
@@ -84,11 +86,13 @@ public class ExampleEchoPipeline {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
+ public void processElement(
+ @Element KV<String, String> element, OutputReceiver<KV<String, String>> receiver)
+ throws Exception {
try {
// Our Library takes a single command in position 0 which it will echo back in the result
SubProcessCommandLineArgs commands = new SubProcessCommandLineArgs();
- Command command = new Command(0, String.valueOf(c.element().getValue()));
+ Command command = new Command(0, String.valueOf(element.getValue()));
commands.putCommand(command);
// The ProcessingKernel deals with the execution of the process
@@ -97,7 +101,7 @@ public class ExampleEchoPipeline {
// Run the command and work through the results
List<String> results = kernel.exec(commands);
for (String s : results) {
- c.output(KV.of(c.element().getKey(), s));
+ receiver.output(KV.of(element.getKey(), s));
}
} catch (Exception ex) {
LOG.error("Error processing element ", ex);
diff --git a/examples/java/src/test/java/org/apache/beam/examples/cookbook/TriggerExampleTest.java b/examples/java/src/test/java/org/apache/beam/examples/cookbook/TriggerExampleTest.java
index 19c83c6eb73..0d9e257fc6c 100644
--- a/examples/java/src/test/java/org/apache/beam/examples/cookbook/TriggerExampleTest.java
+++ b/examples/java/src/test/java/org/apache/beam/examples/cookbook/TriggerExampleTest.java
@@ -29,6 +29,8 @@ import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
@@ -146,8 +148,8 @@ public class TriggerExampleTest {
static class FormatResults extends DoFn<TableRow, String> {
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
- TableRow element = c.element();
+ public void processElement(@Element TableRow element, OutputReceiver<String> receiver)
+ throws Exception {
TableRow row =
new TableRow()
.set("trigger_type", element.get("trigger_type"))
@@ -158,7 +160,7 @@ public class TriggerExampleTest {
.set("isLast", element.get("isLast"))
.set("timing", element.get("timing"))
.set("window", element.get("window"));
- c.output(canonicalFormat(row));
+ receiver.output(canonicalFormat(row));
}
}
}
diff --git a/examples/java/src/test/java/org/apache/beam/examples/subprocess/ExampleEchoPipelineTest.java b/examples/java/src/test/java/org/apache/beam/examples/subprocess/ExampleEchoPipelineTest.java
index 01055d9658a..9981b4040e3 100644
--- a/examples/java/src/test/java/org/apache/beam/examples/subprocess/ExampleEchoPipelineTest.java
+++ b/examples/java/src/test/java/org/apache/beam/examples/subprocess/ExampleEchoPipelineTest.java
@@ -41,6 +41,8 @@ import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
@@ -144,11 +146,13 @@ public class ExampleEchoPipelineTest {
}
@ProcessElement
- public void processElement(ProcessContext c) throws Exception {
+ public void processElement(
+ @Element KV<String, String> element, OutputReceiver<KV<String, String>> receiver)
+ throws Exception {
try {
// Our Library takes a single command in position 0 which it will echo back in the result
SubProcessCommandLineArgs commands = new SubProcessCommandLineArgs();
- Command command = new Command(0, String.valueOf(c.element().getValue()));
+ Command command = new Command(0, String.valueOf(element.getValue()));
commands.putCommand(command);
// The ProcessingKernel deals with the execution of the process
@@ -157,7 +161,7 @@ public class ExampleEchoPipelineTest {
// Run the command and work through the results
List<String> results = kernel.exec(commands);
for (String s : results) {
- c.output(KV.of(c.element().getKey(), s));
+ receiver.output(KV.of(element.getKey(), s));
}
} catch (Exception ex) {
LOG.error("Error processing element ", ex);
diff --git a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/DebuggingWordCount.kt b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/DebuggingWordCount.kt
index 05cb404ac6d..7be4df2c8a9 100644
--- a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/DebuggingWordCount.kt
+++ b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/DebuggingWordCount.kt
@@ -25,6 +25,8 @@ import org.apache.beam.sdk.options.Description
import org.apache.beam.sdk.options.PipelineOptionsFactory
import org.apache.beam.sdk.testing.PAssert
import org.apache.beam.sdk.transforms.DoFn
+import org.apache.beam.sdk.transforms.DoFn.Element
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver
import org.apache.beam.sdk.transforms.ParDo
import org.apache.beam.sdk.values.KV
import org.slf4j.LoggerFactory
@@ -84,18 +86,18 @@ public object DebuggingWordCount {
private val unmatchedWords = Metrics.counter(FilterTextFn::class.java, "unmatchedWords")
@ProcessElement
- fun processElement(c: ProcessContext) {
- if (filter.matcher(c.element().key).matches()) {
+ fun processElement(@Element element: KV<String, Long>, receiver: OutputReceiver<KV<String, Long>>) {
+ if (filter.matcher(element.key).matches()) {
// Log at the "DEBUG" level each element that we match. When executing this pipeline
// these log lines will appear only if the log level is set to "DEBUG" or lower.
- LOG.debug("Matched: ${c.element().key}")
+ LOG.debug("Matched: ${element.key}")
matchedWords.inc()
- c.output(c.element())
+ receiver.output(element)
} else {
// Log at the "TRACE" level each element that is not matched. Different log levels
// can be used to control the verbosity of logging providing an effective mechanism
// to filter less important information.
- LOG.trace("Did not match: ${c.element().key}")
+ LOG.trace("Did not match: ${element.key}")
unmatchedWords.inc()
}
}
diff --git a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/BigQueryTornadoes.kt b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/BigQueryTornadoes.kt
index ec56bc65997..8edc9887235 100644
--- a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/BigQueryTornadoes.kt
+++ b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/BigQueryTornadoes.kt
@@ -27,6 +27,8 @@ import org.apache.beam.sdk.io.gcp.bigquery.WriteResult
import org.apache.beam.sdk.options.*
import org.apache.beam.sdk.transforms.Count
import org.apache.beam.sdk.transforms.DoFn
+import org.apache.beam.sdk.transforms.DoFn.Element
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver
import org.apache.beam.sdk.transforms.PTransform
import org.apache.beam.sdk.transforms.ParDo
import org.apache.beam.sdk.values.KV
@@ -73,10 +75,9 @@ object BigQueryTornadoes {
*/
internal class ExtractTornadoesFn : DoFn<TableRow, Int>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val row = c.element()
+ fun processElement(@Element row: TableRow, receiver: OutputReceiver<Int>) {
if (row["tornado"] as Boolean) {
- c.output(Integer.parseInt(row["month"] as String))
+ receiver.output(Integer.parseInt(row["month"] as String))
}
}
}
@@ -87,11 +88,11 @@ object BigQueryTornadoes {
*/
internal class FormatCountsFn : DoFn<KV<Int, Long>, TableRow>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
+ fun processElement(@Element element: KV<Int, Long>, receiver: OutputReceiver<TableRow>) {
val row = TableRow()
- .set("month", c.element().key)
- .set("tornado_count", c.element().value)
- c.output(row)
+ .set("month", element.key)
+ .set("tornado_count", element.value)
+ receiver.output(row)
}
}
diff --git a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/CombinePerKeyExamples.kt b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/CombinePerKeyExamples.kt
index 9e388031002..4d89b692dd8 100644
--- a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/CombinePerKeyExamples.kt
+++ b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/CombinePerKeyExamples.kt
@@ -26,6 +26,8 @@ import org.apache.beam.sdk.io.gcp.bigquery.WriteResult
import org.apache.beam.sdk.metrics.Metrics
import org.apache.beam.sdk.options.*
import org.apache.beam.sdk.transforms.*
+import org.apache.beam.sdk.transforms.DoFn.Element
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver
import org.apache.beam.sdk.values.KV
import org.apache.beam.sdk.values.PCollection
@@ -70,12 +72,11 @@ object CombinePerKeyExamples {
private val smallerWords = Metrics.counter(ExtractLargeWordsFn::class.java, "smallerWords")
@ProcessElement
- fun processElement(c: ProcessContext) {
- val row = c.element()
+ fun processElement(@Element row: TableRow, receiver: OutputReceiver<KV<String, String>>) {
val playName = row["corpus"] as String
val word = row["word"] as String
if (word.length >= MIN_WORD_LENGTH) {
- c.output(KV.of(word, playName))
+ receiver.output(KV.of(word, playName))
} else {
// Track how many smaller words we're not including. This information will be
// visible in the Monitoring UI.
@@ -90,9 +91,9 @@ object CombinePerKeyExamples {
*/
internal class FormatShakespeareOutputFn : DoFn<KV<String, String>, TableRow>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val row = TableRow().set("word", c.element().key).set("all_plays", c.element().value)
- c.output(row)
+ fun processElement(@Element element: KV<String, String>, receiver: OutputReceiver<TableRow>) {
+ val row = TableRow().set("word", element.key).set("all_plays", element.value)
+ receiver.output(row)
}
}
diff --git a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/FilterExamples.kt b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/FilterExamples.kt
index 2625f5bfec1..21818844896 100644
--- a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/FilterExamples.kt
+++ b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/FilterExamples.kt
@@ -25,6 +25,9 @@ import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult
import org.apache.beam.sdk.options.*
import org.apache.beam.sdk.transforms.*
+import org.apache.beam.sdk.transforms.DoFn.Element
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver
+import org.apache.beam.sdk.transforms.DoFn.SideInput
import org.apache.beam.sdk.values.PCollection
import java.util.logging.Logger
@@ -80,8 +83,7 @@ object FilterExamples {
*/
internal class ProjectionFn : DoFn<TableRow, TableRow>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val row = c.element()
+ fun processElement(@Element row: TableRow, receiver: OutputReceiver<TableRow>) {
// Grab year, month, day, mean_temp from the row
val year = Integer.parseInt(row["year"] as String)
val month = Integer.parseInt(row["month"] as String)
@@ -94,7 +96,7 @@ object FilterExamples {
.set("month", month)
.set("day", day)
.set("mean_temp", meanTemp)
- c.output(outRow)
+ receiver.output(outRow)
}
}
@@ -108,11 +110,10 @@ object FilterExamples {
internal class FilterSingleMonthDataFn(private var monthFilter: Int?) : DoFn<TableRow, TableRow>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val row = c.element()
+ fun processElement(@Element row: TableRow, receiver: OutputReceiver<TableRow>) {
val month = row["month"]
if (month == this.monthFilter) {
- c.output(row)
+ receiver.output(row)
}
}
}
@@ -123,10 +124,9 @@ object FilterExamples {
*/
internal class ExtractTempFn : DoFn<TableRow, Double>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val row = c.element()
+ fun processElement(@Element row: TableRow, receiver: OutputReceiver<Double>) {
val meanTemp = java.lang.Double.parseDouble(row["mean_temp"].toString())
- c.output(meanTemp)
+ receiver.output(meanTemp)
}
}
@@ -158,15 +158,17 @@ object FilterExamples {
ParDo.of(
object : DoFn<TableRow, TableRow>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val meanTemp = java.lang.Double.parseDouble(c.element()["mean_temp"].toString())
- val gTemp = c.sideInput(globalMeanTemp)
+ fun processElement(
+ @SideInput("globalMeanTemp") gTemp: Double,
+ @Element element: TableRow,
+ receiver: OutputReceiver<TableRow>) {
+ val meanTemp = java.lang.Double.parseDouble(element["mean_temp"].toString())
if (meanTemp < gTemp) {
- c.output(c.element())
+ receiver.output(element)
}
}
})
- .withSideInputs(globalMeanTemp))
+ .withSideInput("globalMeanTemp", globalMeanTemp))
}
}
diff --git a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/JoinExamples.kt b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/JoinExamples.kt
index 2f2215e1d96..7f81629ae46 100644
--- a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/JoinExamples.kt
+++ b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/JoinExamples.kt
@@ -26,6 +26,8 @@ import org.apache.beam.sdk.options.PipelineOptions
import org.apache.beam.sdk.options.PipelineOptionsFactory
import org.apache.beam.sdk.options.Validation
import org.apache.beam.sdk.transforms.DoFn
+import org.apache.beam.sdk.transforms.DoFn.Element
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver
import org.apache.beam.sdk.transforms.ParDo
import org.apache.beam.sdk.transforms.join.CoGbkResult
import org.apache.beam.sdk.transforms.join.CoGroupByKey
@@ -89,13 +91,14 @@ object JoinExamples {
ParDo.of(
object : DoFn<KV<String, CoGbkResult>, KV<String, String>>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val e = c.element()
- val countryCode = e.key
- val countryName = e.value.getOnly(countryInfoTag)
- for (ei in c.element().value.getAll(eventInfoTag)) {
+ fun processElement(
+ @Element element: KV<String, CoGbkResult>,
+ receiver: OutputReceiver<KV<String, String>>) {
+ val countryCode = element.key
+ val countryName = element.value.getOnly(countryInfoTag)
+ for (ei in element.value.getAll(eventInfoTag)) {
// Generate a string that combines information from both collection values
- c.output(
+ receiver.output(
KV.of<String, String>(
countryCode,
"Country name: $countryName, Event info: $ei"))
@@ -109,9 +112,9 @@ object JoinExamples {
ParDo.of(
object : DoFn<KV<String, String>, String>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val outputString = "Country code: ${c.element().key}, ${c.element().value}"
- c.output(outputString)
+ fun processElement(@Element element: KV<String, String>, receiver: OutputReceiver<String>) {
+ val outputString = "Country code: ${element.key}, ${element.value}"
+ receiver.output(outputString)
}
}))
}
@@ -122,14 +125,13 @@ object JoinExamples {
*/
internal class ExtractEventDataFn : DoFn<TableRow, KV<String, String>>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val row = c.element()
+ fun processElement(@Element row: TableRow, receiver: OutputReceiver<KV<String, String>>) {
val countryCode = row["ActionGeo_CountryCode"] as String
val sqlDate = row["SQLDATE"] as String
val actor1Name = row["Actor1Name"] as String
val sourceUrl = row["SOURCEURL"] as String
val eventInfo = "Date: $sqlDate, Actor1: $actor1Name, url: $sourceUrl"
- c.output(KV.of(countryCode, eventInfo))
+ receiver.output(KV.of(countryCode, eventInfo))
}
}
@@ -139,11 +141,10 @@ object JoinExamples {
*/
internal class ExtractCountryInfoFn : DoFn<TableRow, KV<String, String>>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val row = c.element()
+ fun processElement(@Element row: TableRow, receiver: OutputReceiver<KV<String, String>>) {
val countryCode = row["FIPSCC"] as String
val countryName = row["HumanName"] as String
- c.output(KV.of(countryCode, countryName))
+ receiver.output(KV.of(countryCode, countryName))
}
}
diff --git a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/MaxPerKeyExamples.kt b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/MaxPerKeyExamples.kt
index 11418d3933c..ad16ecde965 100644
--- a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/MaxPerKeyExamples.kt
+++ b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/MaxPerKeyExamples.kt
@@ -26,6 +26,8 @@ import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult
import org.apache.beam.sdk.options.*
import org.apache.beam.sdk.transforms.DoFn
+import org.apache.beam.sdk.transforms.DoFn.Element
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver
import org.apache.beam.sdk.transforms.Max
import org.apache.beam.sdk.transforms.PTransform
import org.apache.beam.sdk.transforms.ParDo
@@ -73,22 +75,21 @@ object MaxPerKeyExamples {
*/
internal class ExtractTempFn : DoFn<TableRow, KV<Int, Double>>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val row = c.element()
+ fun processElement(@Element row: TableRow, receiver: OutputReceiver<KV<Int, Double>>) {
val month = Integer.parseInt(row["month"] as String)
val meanTemp = java.lang.Double.parseDouble(row["mean_temp"].toString())
- c.output(KV.of(month, meanTemp))
+ receiver.output(KV.of(month, meanTemp))
}
}
/** Format the results to a TableRow, to save to BigQuery. */
internal class FormatMaxesFn : DoFn<KV<Int, Double>, TableRow>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
+ fun processElement(@Element element: KV<Int, Double>, receiver: OutputReceiver<TableRow>) {
val row = TableRow()
- .set("month", c.element().key)
- .set("max_mean_temp", c.element().value)
- c.output(row)
+ .set("month", element.key)
+ .set("max_mean_temp", element.value)
+ receiver.output(row)
}
}
diff --git a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/TriggerExample.kt b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/TriggerExample.kt
index 4afa7d0dfc7..bb8c0900e31 100644
--- a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/TriggerExample.kt
+++ b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/cookbook/TriggerExample.kt
@@ -34,6 +34,9 @@ import org.apache.beam.sdk.options.Description
import org.apache.beam.sdk.options.PipelineOptionsFactory
import org.apache.beam.sdk.options.StreamingOptions
import org.apache.beam.sdk.transforms.DoFn
+import org.apache.beam.sdk.transforms.DoFn.Element
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver
+import org.apache.beam.sdk.transforms.DoFn.Timestamp
import org.apache.beam.sdk.transforms.GroupByKey
import org.apache.beam.sdk.transforms.PTransform
import org.apache.beam.sdk.transforms.ParDo
@@ -364,15 +367,17 @@ object TriggerExample {
@ProcessElement
@Throws(Exception::class)
- fun processElement(c: ProcessContext) {
- val flows = c.element().value
+ fun processElement(
+ @Element element: KV<String, Iterable<Int>>,
+ receiver: OutputReceiver<KV<String, String>>) {
+ val flows = element.value
var sum = 0
var numberOfRecords = 0L
for (value in flows) {
sum += value
numberOfRecords++
}
- c.output(KV.of<String, String>(c.element().key, "$sum,$numberOfRecords"))
+ receiver.output(KV.of<String, String>(element.key, "$sum,$numberOfRecords"))
}
}))
return results.apply(ParDo.of(FormatTotalFlow(triggerType)))
@@ -387,20 +392,25 @@ object TriggerExample {
@ProcessElement
@Throws(Exception::class)
- fun processElement(c: ProcessContext, window: BoundedWindow) {
- val values = c.element().value.split(",".toRegex()).toTypedArray()
+ fun processElement(
+ @Element element: KV<String, String>,
+ window: BoundedWindow,
+ pane: PaneInfo,
+ @Timestamp timestamp: Instant,
+ receiver: OutputReceiver<TableRow>) {
+ val values = element.value.split(",".toRegex()).toTypedArray()
val row = TableRow()
.set("trigger_type", triggerType)
- .set("freeway", c.element().key)
+ .set("freeway", element.key)
.set("total_flow", Integer.parseInt(values[0]))
.set("number_of_records", java.lang.Long.parseLong(values[1]))
.set("window", window.toString())
- .set("isFirst", c.pane().isFirst)
- .set("isLast", c.pane().isLast)
- .set("timing", c.pane().timing.toString())
- .set("event_time", c.timestamp().toString())
+ .set("isFirst", pane.isFirst)
+ .set("isLast", pane.isLast)
+ .set("timing", pane.timing.toString())
+ .set("event_time", timestamp.toString())
.set("processing_time", Instant.now().toString())
- c.output(row)
+ receiver.output(row)
}
}
@@ -412,8 +422,8 @@ object TriggerExample {
@ProcessElement
@Throws(Exception::class)
- fun processElement(c: ProcessContext) {
- val laneInfo = c.element().split(",".toRegex()).toTypedArray()
+ fun processElement(@Element element: String, receiver: OutputReceiver<KV<String, Int>>) {
+ val laneInfo = element.split(",".toRegex()).toTypedArray()
if ("timestamp" == laneInfo[0]) {
// Header row
return
@@ -429,7 +439,7 @@ object TriggerExample {
if (totalFlow == null || totalFlow <= 0) {
return
}
- c.output(KV.of<String, Int>(freeway, totalFlow))
+ receiver.output(KV.of<String, Int>(freeway, totalFlow))
}
companion object {
@@ -486,7 +496,7 @@ object TriggerExample {
@ProcessElement
@Throws(Exception::class)
- fun processElement(c: ProcessContext) {
+ fun processElement(@Element element: String, receiver: OutputReceiver<String>) {
var timestamp = Instant.now()
if (random.nextDouble() < THRESHOLD) {
val range = MAX_DELAY - MIN_DELAY
@@ -494,7 +504,7 @@ object TriggerExample {
val delayInMillis = TimeUnit.MINUTES.toMillis(delayInMinutes.toLong())
timestamp = Instant(timestamp.millis - delayInMillis)
}
- c.outputWithTimestamp(c.element(), timestamp)
+ receiver.outputWithTimestamp(element, timestamp)
}
companion object {
diff --git a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/snippets/Snippets.kt b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/snippets/Snippets.kt
index d2f58c215a5..eb209937248 100644
--- a/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/snippets/Snippets.kt
+++ b/examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/snippets/Snippets.kt
@@ -33,6 +33,8 @@ import org.apache.beam.sdk.io.gcp.bigquery.DynamicDestinations
import org.apache.beam.sdk.io.gcp.bigquery.TableDestination
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult
import org.apache.beam.sdk.transforms.*
+import org.apache.beam.sdk.transforms.DoFn.Element
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver
import org.apache.beam.sdk.transforms.join.CoGbkResult
import org.apache.beam.sdk.transforms.join.CoGroupByKey
import org.apache.beam.sdk.transforms.join.KeyedPCollectionTuple
@@ -360,13 +362,12 @@ object Snippets {
ParDo.of(
object : DoFn<KV<String, CoGbkResult>, String>() {
@ProcessElement
- fun processElement(c: ProcessContext) {
- val e = c.element()
- val name = e.key
- val emailsIter = e.value.getAll(emailsTag)
- val phonesIter = e.value.getAll(phonesTag)
+ fun processElement(@Element element: KV<String, CoGbkResult>, receiver: OutputReceiver<String>) {
+ val name = element.key
+ val emailsIter = element.value.getAll(emailsTag)
+ val phonesIter = element.value.getAll(phonesTag)
val formattedResult = formatCoGbkResults(name, emailsIter, phonesIter)
- c.output(formattedResult)
+ receiver.output(formattedResult)
}
}))
}