|
| 1 | +--- |
| 2 | +# Licensed to the Apache Software Foundation (ASF) under one |
| 3 | +# or more contributor license agreements. See the NOTICE file |
| 4 | +# distributed with this work for additional information |
| 5 | +# regarding copyright ownership. The ASF licenses this file |
| 6 | +# to you under the Apache License, Version 2.0 (the |
| 7 | +# "License"); you may not use this file except in compliance |
| 8 | +# with the License. You may obtain a copy of the License at |
| 9 | +# |
| 10 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | +# |
| 12 | +# Unless required by applicable law or agreed to in writing, |
| 13 | +# software distributed under the License is distributed on an |
| 14 | +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 15 | +# KIND, either express or implied. See the License for the |
| 16 | +# specific language governing permissions and limitations |
| 17 | +# under the License. |
| 18 | +--- |
| 19 | +name: beam-dofn-modernizer |
| 20 | +description: Rewrite Apache Beam DoFn methods (@ProcessElement, @OnTimer, @OnWindowExpiration) to remove legacy ProcessContext or OnTimerContext usage. Use this skill when you encounter DoFn methods that use context.element(), context.output(), etc., and need to modernize them using parameter injection (@Element, @Timestamp, @Pane, OutputReceiver, MultiOutputReceiver). |
| 21 | +--- |
| 22 | + |
| 23 | +# Modernizing Apache Beam DoFns |
| 24 | + |
| 25 | +Apache Beam has moved towards parameter injection in `DoFn` methods to improve readability and allow for more efficient execution. This skill helps you migrate legacy `ProcessContext` and `OnTimerContext` usage to modern annotated parameters. |
| 26 | + |
| 27 | +## Core Mappings |
| 28 | + |
| 29 | +When rewriting a `@ProcessElement` or `@OnTimer` method, replace the context argument with the corresponding parameters based on the usage: |
| 30 | + |
| 31 | +| Legacy Context Usage (e.g. `ProcessContext c`) | Modern Parameter Replacement | |
| 32 | +| :--- | :--- | |
| 33 | +| `c.element()` | `@Element T element` | |
| 34 | +| `c.timestamp()` | `@Timestamp Instant timestamp` | |
| 35 | +| `c.pane()` | `PaneInfo pane` | |
| 36 | +| `c.window()` | `BoundedWindow window` | |
| 37 | +| `c.sideInput(PCollectionView<T> view)` | `@SideInput("viewName") T value` | |
| 38 | +| `c.getPipelineOptions()` | `PipelineOptions options` | |
| 39 | +| `c.output(value)` | `OutputReceiver<T> receiver` then `receiver.output(value)` | |
| 40 | +| `c.output(tag, value)` | `MultiOutputReceiver receiver` then `receiver.get(tag).output(value)` | |
| 41 | +| `c.outputWithTimestamp(value, ts)` | `OutputReceiver<T> receiver` then `receiver.outputWithTimestamp(value, ts)` | |
| 42 | + |
| 43 | +## Method Signature Changes |
| 44 | + |
| 45 | +### @ProcessElement |
| 46 | + |
| 47 | +**Legacy:** |
| 48 | +```java |
| 49 | +@ProcessElement |
| 50 | +public void processElement(ProcessContext c) { |
| 51 | + T element = c.element(); |
| 52 | + c.output(transform(element)); |
| 53 | +} |
| 54 | +``` |
| 55 | + |
| 56 | +**Modern:** |
| 57 | +```java |
| 58 | +@ProcessElement |
| 59 | +public void processElement( |
| 60 | + @Element T element, |
| 61 | + @Timestamp Instant timestamp, |
| 62 | + OutputReceiver<V> receiver) { |
| 63 | + receiver.output(transform(element)); |
| 64 | +} |
| 65 | +``` |
| 66 | + |
| 67 | +### @OnTimer |
| 68 | + |
| 69 | +**Legacy:** |
| 70 | +```java |
| 71 | +@OnTimer("timerId") |
| 72 | +public void onTimer(OnTimerContext c) { |
| 73 | + c.output(someValue); |
| 74 | +} |
| 75 | +``` |
| 76 | + |
| 77 | +**Modern:** |
| 78 | +```java |
| 79 | +@OnTimer("timerId") |
| 80 | +public void onTimer( |
| 81 | + @Timestamp Instant timestamp, |
| 82 | + BoundedWindow window, |
| 83 | + OutputReceiver<V> receiver) { |
| 84 | + receiver.output(someValue); |
| 85 | +} |
| 86 | +``` |
| 87 | + |
| 88 | +## Best Practices |
| 89 | + |
| 90 | +1. **Specific OutputReceiver**: If the method only outputs to the main output, use `OutputReceiver<T>`. If it outputs to multiple tags, use `MultiOutputReceiver`. |
| 91 | +2. **Element Type**: Ensure the `@Element` parameter type matches the input type of the `DoFn`. |
| 92 | +3. **Imports**: Don't forget to add imports for: |
| 93 | + * `org.apache.beam.sdk.transforms.DoFn.Element` |
| 94 | + * `org.apache.beam.sdk.transforms.DoFn.Timestamp` |
| 95 | + * `org.apache.beam.sdk.transforms.DoFn.OutputReceiver` |
| 96 | + * `org.apache.beam.sdk.transforms.DoFn.MultiOutputReceiver` (if needed) |
| 97 | + * `org.apache.beam.sdk.values.PCollectionView` (if using `@SideInput`) |
| 98 | + * `org.apache.beam.sdk.transforms.DoFn.SideInput` |
| 99 | + * `org.apache.beam.sdk.transforms.windowing.PaneInfo` |
| 100 | +4. **Side Inputs**: When using `@SideInput`, make sure to use the correct name that matches the one passed to `ParDo.withSideInput("name", view)`. |
| 101 | +5. **Parameter Naming and Redundant Variables**: Use descriptive names for the `@Element` parameter (e.g., `record`, `line`, `row`) instead of a generic `element` if it improves readability. Do not create a redundant local variable to copy the element (e.g., `MyType elm = element;`), use the parameter directly. |
| 102 | + |
| 103 | +## Example Conversion |
| 104 | + |
| 105 | +### Before: |
| 106 | +```java |
| 107 | +@ProcessElement |
| 108 | +public void processElement(ProcessContext c) { |
| 109 | + KV<String, Integer> element = c.element(); |
| 110 | + Instant ts = c.timestamp(); |
| 111 | + if (element.getValue() > threshold) { |
| 112 | + c.output(element.getKey()); |
| 113 | + c.output(specialTag, element.getValue()); |
| 114 | + } |
| 115 | +} |
| 116 | +``` |
| 117 | + |
| 118 | +### After: |
| 119 | +```java |
| 120 | +@ProcessElement |
| 121 | +public void processElement( |
| 122 | + @Element KV<String, Integer> element, |
| 123 | + @Timestamp Instant timestamp, |
| 124 | + MultiOutputReceiver receiver) { |
| 125 | + if (element.getValue() > threshold) { |
| 126 | + receiver.get(mainTag).output(element.getKey()); |
| 127 | + receiver.get(specialTag).output(element.getValue()); |
| 128 | + } |
| 129 | +} |
| 130 | +``` |
| 131 | +> [!NOTE] |
| 132 | +> If you only have one output, use `OutputReceiver<String> receiver` and `receiver.output(element.getKey())`. |
| 133 | +
|
| 134 | +## Side Input Conversion |
| 135 | + |
| 136 | +Modernizing side inputs involves removing the `PCollectionView` from the `DoFn` constructor and using `@SideInput` parameter injection instead. |
| 137 | + |
| 138 | +### Before (Legacy): |
| 139 | + |
| 140 | +**PTransform/Pipeline side:** |
| 141 | +```java |
| 142 | +PCollectionView<String> myView = ...; |
| 143 | +input.apply(ParDo.of(new MyFn(myView)).withSideInputs(myView)); |
| 144 | +``` |
| 145 | + |
| 146 | +**DoFn side:** |
| 147 | +```java |
| 148 | +class MyFn extends DoFn<T, V> { |
| 149 | + private final PCollectionView<String> view; |
| 150 | + MyFn(PCollectionView<String> view) { this.view = view; } |
| 151 | + |
| 152 | + @ProcessElement |
| 153 | + public void processElement(ProcessContext c) { |
| 154 | + String value = c.sideInput(view); |
| 155 | + // ... |
| 156 | + } |
| 157 | +} |
| 158 | +``` |
| 159 | + |
| 160 | +### Nullable Side Inputs |
| 161 | + |
| 162 | +If a side input is optional and a `DoFn` has conditional logic based on whether the side input is present, it is best to split the `DoFn` into two separate classes: one that requires the side input and one that does not. This avoids creating complex, conditional `DoFn`s and ensures type safety. |
| 163 | + |
| 164 | +**PTransform/Pipeline side:** |
| 165 | +```java |
| 166 | +PCollectionView<String> myView = ...; |
| 167 | +input.apply(ParDo.of(new MyFn(myView)).withSideInputs(myView)); |
| 168 | +//or |
| 169 | +input.apply(ParDo.of(new MyFn(null))); // to introduce null |
| 170 | +``` |
| 171 | +**DoFn side:** |
| 172 | +```java |
| 173 | +class MyFn extends DoFn<T, V> { |
| 174 | + private final PCollectionView<String> view; |
| 175 | + MyFn(PCollectionView<String> view) { this.view = view; } |
| 176 | + |
| 177 | + @ProcessElement |
| 178 | + public void processElement(ProcessContext c) { |
| 179 | + String value = null; |
| 180 | + if (this.view != null) { // can do conditional side input |
| 181 | + value = c.sideInput(this.view); |
| 182 | + } |
| 183 | + |
| 184 | + // ... |
| 185 | + } |
| 186 | +} |
| 187 | +``` |
| 188 | + |
| 189 | +### After (Modern): |
| 190 | + |
| 191 | +**PTransform/Pipeline side:** |
| 192 | +```java |
| 193 | +PCollectionView<String> myView = ...; |
| 194 | +input.apply(ParDo.of(new MyFn()).withSideInput("sideInputName", myView)); |
| 195 | +``` |
| 196 | + |
| 197 | +**DoFn side:** |
| 198 | +```java |
| 199 | +class MyFn extends DoFn<T, V> { |
| 200 | + @ProcessElement |
| 201 | + public void processElement( |
| 202 | + @Element T element, |
| 203 | + @SideInput("sideInputName") String value) { |
| 204 | + // value is injected directly |
| 205 | + } |
| 206 | +} |
| 207 | +``` |
0 commit comments