olehborysevych commented on code in PR #25253: URL: https://github.com/apache/beam/pull/25253#discussion_r1169794761
########## learning/tour-of-beam/learning-content/core-transforms/additional-outputs/description.md: ########## @@ -0,0 +1,322 @@ +<!-- +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at +http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +--> +# Additional outputs +{{if (eq .Sdk "go")}} +While `beam.ParDo` always produces an output `PCollection`, your `DoFn` can produce any number of additional output `PCollection`s, or even none at all. If you choose to have multiple outputs, your `DoFn` needs to be called with the `ParDo` function that matches the number of outputs. `beam.ParDo2` for two output `PCollection`s, `beam.ParDo3` for three and so on until `beam.ParDo7`. If you need more, you can use `beam.ParDoN` which will return a `[]beam.PCollection`. + +### Tags for multiple outputs + +The Go SDK doesn’t use output tags, and instead uses positional ordering for multiple output `PCollection`s. + +``` +// beam.ParDo3 returns PCollections in the same order as +// the emit function parameters in processWords. +below, above, marked := beam.ParDo3(s, processWords, words) + +// processWordsMixed uses both a standard return and an emitter function. +// The standard return produces the first PCollection from beam.ParDo2, +// and the emitter produces the second PCollection. +length, mixedMarked := beam.ParDo2(s, processWordsMixed, words) +``` + +### Emitting to multiple outputs in your DoFn + +Call emitter functions as needed to produce 0 or more elements for its matching `PCollection`. The same value can be emitted with multiple emitters. As normal, do not mutate values after emitting them from any emitter. + +All emitters should be registered using a generic `register.EmitterX[...]` function. This optimizes runtime execution of the emitter. + +`DoFn`s can also return a single element via the standard return. The standard return is always the first PCollection returned from `beam.ParDo`. Other emitters output to their own `PCollection`s in their defined parameter order. + +``` +// processWords is a DoFn that has 3 output PCollections. The emitter functions +// are matched in positional order to the PCollections returned by beam.ParDo3. +func processWords(word string, emitBelowCutoff, emitAboveCutoff, emitMarked func(string)) { + const cutOff = 5 + if len(word) < cutOff { + emitBelowCutoff(word) + } else { + emitAboveCutoff(word) + } + if isMarkedWord(word) { + emitMarked(word) + } +} +``` + +### Accessing additional parameters in your DoFn + +In addition to the element, Beam will populate other parameters to your DoFn’s `ProcessElement` method. Any combination of these parameters can be added to your process method in a standard order. + +**context.Context**: To support consolidated logging and user defined metrics, a `context.Context` parameter can be requested. Per Go conventions, if present it’s required to be the first parameter of the `DoFn` method. + +``` +func MyDoFn(ctx context.Context, word string) string { ... } +``` + +**Timestamp**: To access the timestamp of an input element, add a `beam.EventTime` parameter before the element. For example: + +``` +func MyDoFn(ts beam.EventTime, word string) string { ... } +``` + +**Window**: To access the window an input element falls into, add a `beam.Window` parameter before the element. If an element falls in multiple windows (for example, this will happen when using `SlidingWindows`), then the `ProcessElement` method will be invoked multiple time for the element, once for each window. Since `beam.Window` is an interface it’s possible to type assert to the concrete implementation of the window. For example, when fixed windows are being used, the window is of type `window.IntervalWindow`. + +``` +func MyDoFn(w beam.Window, word string) string { + iw := w.(window.IntervalWindow) + ... +} +``` + +**PaneInfo**: When triggers are used, Beam provides `beam.PaneInfo` object that contains information about the current firing. Using `beam.PaneInfo` you can determine whether this is an early or a late firing, and how many times this window has already fired for this key. + +``` +func extractWordsFn(pn beam.PaneInfo, line string, emitWords func(string)) { + if pn.Timing == typex.PaneEarly || pn.Timing == typex.PaneOnTime { + // ... perform operation ... + } + if pn.Timing == typex.PaneLate { + // ... perform operation ... + } + if pn.IsFirst { + // ... perform operation ... + } + if pn.IsLast { + // ... perform operation ... + } + + words := strings.Split(line, " ") + for _, w := range words { + emitWords(w) + } +} +``` +{{end}} +{{if (eq .Sdk "java")}} +While `ParDo` always outputs the main output of `PCollection` (as a return value from apply), you can also force your `ParDo` to output any number of additional `PCollection` outputs. If you decide to have multiple outputs, your `ParDo` will return all the `PCollection` output (including the main output) combined. This will be useful when you are working with big data or a database that needs to be divided into different collections. You get a combined `PCollectionTuple`, you can use `TupleTag` to get a `PCollection`. + +A `PCollectionTuple` is an immutable tuple of heterogeneously typed `PCollection`, "with keys" `TupleTags`. A `PCollectionTuple` can be used as input or output for `PTransform` receiving or creating multiple `PCollection` inputs or outputs, which can be of different types, for example, `ParDo` with multiple outputs. + +A `TupleTag` is a typed tag used as the key of a heterogeneously typed tuple, for example `PCollectionTuple`. Its general type parameter allows you to track the static type of things stored in tuples. + +### Tags for multiple outputs + +``` + .of(new DoFn<String, String>() { Review Comment: Hey @nausharipov could you please check if this markdown issue is caused by some flutter issue or we can fix this? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
