Skip to content

Commit e530809

Browse files
AbacnAbacn
authored
Revert "remove processContext usage across examples (java and kotlin)" (#38341)
* Revert "remove processContext usage across examples (java and kotlin) (#37937)" This reverts commit 4151dde. * Restore SKILLs.md as it does not affect Beam playground
1 parent 9a5318e commit e530809

78 files changed

Lines changed: 547 additions & 855 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/IcebergBatchWriteExample.java‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,6 @@
2828
import org.apache.beam.sdk.options.Validation;
2929
import org.apache.beam.sdk.schemas.Schema;
3030
import org.apache.beam.sdk.transforms.DoFn;
31-
import org.apache.beam.sdk.transforms.DoFn.Element;
32-
import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
3331
import org.apache.beam.sdk.transforms.MapElements;
3432
import org.apache.beam.sdk.transforms.PTransform;
3533
import org.apache.beam.sdk.transforms.ParDo;
@@ -76,8 +74,9 @@ private static Row flattenAnalyticsRow(Row row) {
7674

7775
static class ExtractBrowserTransactionsFn extends DoFn<Row, KV<String, Long>> {
7876
@ProcessElement
79-
public void processElement(@Element Row row, OutputReceiver<KV<String, Long>> receiver) {
80-
receiver.output(
77+
public void processElement(ProcessContext c) {
78+
Row row = c.element();
79+
c.output(
8180
KV.of(
8281
Preconditions.checkStateNotNull(row.getString("browser")),
8382
Preconditions.checkStateNotNull(row.getInt64("transactions"))));
@@ -86,13 +85,13 @@ public void processElement(@Element Row row, OutputReceiver<KV<String, Long>> re
8685

8786
static class FormatCountsFn extends DoFn<KV<String, Long>, Row> {
8887
@ProcessElement
89-
public void processElement(@Element KV<String, Long> element, OutputReceiver<Row> receiver) {
88+
public void processElement(ProcessContext c) {
9089
Row row =
9190
Row.withSchema(AGGREGATED_SCHEMA)
92-
.withFieldValue("browser", element.getKey())
93-
.withFieldValue("transaction_count", element.getValue())
91+
.withFieldValue("browser", c.element().getKey())
92+
.withFieldValue("transaction_count", c.element().getValue())
9493
.build();
95-
receiver.output(row);
94+
c.output(row);
9695
}
9796
}
9897

‎examples/java/sql/src/main/java/org/apache/beam/examples/SchemaTransformExample.java‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -42,8 +42,6 @@
4242
import org.apache.beam.sdk.schemas.transforms.Select;
4343
import org.apache.beam.sdk.transforms.Create;
4444
import org.apache.beam.sdk.transforms.DoFn;
45-
import org.apache.beam.sdk.transforms.DoFn.Element;
46-
import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
4745
import org.apache.beam.sdk.transforms.Max;
4846
import org.apache.beam.sdk.transforms.Min;
4947
import org.apache.beam.sdk.transforms.ParDo;
@@ -103,9 +101,9 @@ public LogOutput(String prefix) {
103101
}
104102

105103
@ProcessElement
106-
public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
107-
LOG.info("{}{}", prefix, element);
108-
receiver.output(element);
104+
public void processElement(ProcessContext c) throws Exception {
105+
LOG.info("{}{}", prefix, c.element());
106+
c.output(c.element());
109107
}
110108
}
111109
}

‎examples/java/sql/src/main/java/org/apache/beam/examples/SqlTransformExample.java‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -41,8 +41,6 @@
4141
import org.apache.beam.sdk.schemas.Schema;
4242
import org.apache.beam.sdk.transforms.Create;
4343
import org.apache.beam.sdk.transforms.DoFn;
44-
import org.apache.beam.sdk.transforms.DoFn.Element;
45-
import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
4644
import org.apache.beam.sdk.transforms.ParDo;
4745
import org.apache.beam.sdk.values.PCollection;
4846
import org.apache.beam.sdk.values.Row;
@@ -97,9 +95,9 @@ public LogOutput(String prefix) {
9795
}
9896

9997
@ProcessElement
100-
public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
101-
LOG.info("{}{}", prefix, element);
102-
receiver.output(element);
98+
public void processElement(ProcessContext c) throws Exception {
99+
LOG.info("{}{}", prefix, c.element());
100+
c.output(c.element());
103101
}
104102
}
105103
}

‎examples/java/src/main/java/org/apache/beam/examples/ApproximateQuantilesExample.java‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,8 +24,6 @@
2424
import org.apache.beam.sdk.transforms.ApproximateQuantiles;
2525
import org.apache.beam.sdk.transforms.Create;
2626
import org.apache.beam.sdk.transforms.DoFn;
27-
import org.apache.beam.sdk.transforms.DoFn.Element;
28-
import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
2927
import org.apache.beam.sdk.transforms.ParDo;
3028
import org.apache.beam.sdk.values.PCollection;
3129
import org.slf4j.Logger;
@@ -72,9 +70,9 @@ public LogOutput(String prefix) {
7270
}
7371

7472
@ProcessElement
75-
public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
76-
LOG.info("{}{}", prefix, element);
77-
receiver.output(element);
73+
public void processElement(ProcessContext c) throws Exception {
74+
LOG.info("{}{}", prefix, c.element());
75+
c.output(c.element());
7876
}
7977
}
8078
}

‎examples/java/src/main/java/org/apache/beam/examples/CoCombineTransformExample.java‎

Lines changed: 6 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -46,8 +46,6 @@
4646
import org.apache.beam.sdk.transforms.CombineFns;
4747
import org.apache.beam.sdk.transforms.Create;
4848
import org.apache.beam.sdk.transforms.DoFn;
49-
import org.apache.beam.sdk.transforms.DoFn.Element;
50-
import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
5149
import org.apache.beam.sdk.transforms.Max;
5250
import org.apache.beam.sdk.transforms.Min;
5351
import org.apache.beam.sdk.transforms.ParDo;
@@ -187,16 +185,13 @@ public Long apply(Long input) {
187185
new DoFn<
188186
KV<Long, CombineFns.CoCombineResult>, KV<Long, Iterable<KV<String, Long>>>>() {
189187
@ProcessElement
190-
public void processElement(
191-
@Element KV<Long, CombineFns.CoCombineResult> element,
192-
OutputReceiver<KV<Long, Iterable<KV<String, Long>>>> receiver)
193-
throws Exception {
194-
CombineFns.CoCombineResult e = element.getValue();
188+
public void processElement(ProcessContext c) throws Exception {
189+
CombineFns.CoCombineResult e = c.element().getValue();
195190
ArrayList<KV<String, Long>> o = new ArrayList<KV<String, Long>>();
196191
o.add(KV.of(minTag.getId(), e.get(minTag)));
197192
o.add(KV.of(maxTag.getId(), e.get(maxTag)));
198193
o.add(KV.of(sumTag.getId(), e.get(sumTag)));
199-
receiver.output(KV.of(element.getKey(), o));
194+
c.output(KV.of(c.element().getKey(), o));
200195
}
201196
}));
202197

@@ -215,9 +210,9 @@ public LogOutput(String prefix) {
215210
}
216211

217212
@ProcessElement
218-
public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
219-
LOG.info("{}{}", prefix, element);
220-
receiver.output(element);
213+
public void processElement(ProcessContext c) throws Exception {
214+
LOG.info("{}{}", prefix, c.element());
215+
c.output(c.element());
221216
}
222217
}
223218
}

‎examples/java/src/main/java/org/apache/beam/examples/CoGroupByKeyExample.java‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,6 @@
2222
import org.apache.beam.sdk.options.PipelineOptionsFactory;
2323
import org.apache.beam.sdk.transforms.Create;
2424
import org.apache.beam.sdk.transforms.DoFn;
25-
import org.apache.beam.sdk.transforms.DoFn.Element;
26-
import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
2725
import org.apache.beam.sdk.transforms.ParDo;
2826
import org.apache.beam.sdk.transforms.join.CoGbkResult;
2927
import org.apache.beam.sdk.transforms.join.CoGroupByKey;
@@ -86,9 +84,9 @@ public LogOutput(String prefix) {
8684
}
8785

8886
@ProcessElement
89-
public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
90-
LOG.info("{}{}", prefix, element);
91-
receiver.output(element);
87+
public void processElement(ProcessContext c) throws Exception {
88+
LOG.info("{}{}", prefix, c.element());
89+
c.output(c.element());
9290
}
9391
}
9492
}

‎examples/java/src/main/java/org/apache/beam/examples/CombineExample.java‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,6 @@
2323
import org.apache.beam.sdk.transforms.Combine;
2424
import org.apache.beam.sdk.transforms.Create;
2525
import org.apache.beam.sdk.transforms.DoFn;
26-
import org.apache.beam.sdk.transforms.DoFn.Element;
27-
import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
2826
import org.apache.beam.sdk.transforms.ParDo;
2927
import org.apache.beam.sdk.transforms.Sum;
3028
import org.apache.beam.sdk.values.PCollection;
@@ -70,9 +68,9 @@ public LogOutput(String prefix) {
7068
}
7169

7270
@ProcessElement
73-
public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
74-
LOG.info("{}{}", prefix, element);
75-
receiver.output(element);
71+
public void processElement(ProcessContext c) throws Exception {
72+
LOG.info("{}{}", prefix, c.element());
73+
c.output(c.element());
7674
}
7775
}
7876
}

‎examples/java/src/main/java/org/apache/beam/examples/CountExample.java‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,6 @@
2323
import org.apache.beam.sdk.transforms.Count;
2424
import org.apache.beam.sdk.transforms.Create;
2525
import org.apache.beam.sdk.transforms.DoFn;
26-
import org.apache.beam.sdk.transforms.DoFn.Element;
27-
import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
2826
import org.apache.beam.sdk.transforms.ParDo;
2927
import org.apache.beam.sdk.values.PCollection;
3028
import org.slf4j.Logger;
@@ -65,9 +63,9 @@ public LogOutput(String prefix) {
6563
}
6664

6765
@ProcessElement
68-
public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
69-
LOG.info("{}{}", prefix, element);
70-
receiver.output(element);
66+
public void processElement(ProcessContext c) throws Exception {
67+
LOG.info("{}{}", prefix, c.element());
68+
c.output(c.element());
7169
}
7270
}
7371
}

‎examples/java/src/main/java/org/apache/beam/examples/CountPerKeyExample.java‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,6 @@
2323
import org.apache.beam.sdk.transforms.Count;
2424
import org.apache.beam.sdk.transforms.Create;
2525
import org.apache.beam.sdk.transforms.DoFn;
26-
import org.apache.beam.sdk.transforms.DoFn.Element;
27-
import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
2826
import org.apache.beam.sdk.transforms.ParDo;
2927
import org.apache.beam.sdk.values.KV;
3028
import org.apache.beam.sdk.values.PCollection;
@@ -69,9 +67,9 @@ public LogOutput(String prefix) {
6967
}
7068

7169
@ProcessElement
72-
public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
73-
LOG.info("{}{}", prefix, element);
74-
receiver.output(element);
70+
public void processElement(ProcessContext c) throws Exception {
71+
LOG.info("{}{}", prefix, c.element());
72+
c.output(c.element());
7573
}
7674
}
7775
}

‎examples/java/src/main/java/org/apache/beam/examples/CreateExample.java‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -27,8 +27,6 @@
2727
import org.apache.beam.sdk.options.PipelineOptionsFactory;
2828
import org.apache.beam.sdk.transforms.Create;
2929
import org.apache.beam.sdk.transforms.DoFn;
30-
import org.apache.beam.sdk.transforms.DoFn.Element;
31-
import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
3230
import org.apache.beam.sdk.transforms.ParDo;
3331
import org.apache.beam.sdk.values.KV;
3432
import org.apache.beam.sdk.values.PCollection;
@@ -81,9 +79,9 @@ public LogOutput(String prefix) {
8179
}
8280

8381
@ProcessElement
84-
public void processElement(@Element T element, OutputReceiver<T> receiver) throws Exception {
85-
LOG.info("{}{}", prefix, element);
86-
receiver.output(element);
82+
public void processElement(ProcessContext c) throws Exception {
83+
LOG.info("{}{}", prefix, c.element());
84+
c.output(c.element());
8785
}
8886
}
8987
}

0 commit comments

Comments
 (0)