This is an automated email from the ASF dual-hosted git repository. pabloem 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 77cf84c [Beam 7447] Add Snippets for Options pattern. (#8710) 77cf84c is described below commit 77cf84c634381495d45a112a9d147ad69394c0d4 Author: Reza Rokni <7542791+rezaro...@users.noreply.github.com> AuthorDate: Fri May 31 06:56:52 2019 +0800 [Beam 7447] Add Snippets for Options pattern. (#8710) * BEAM-7447 Add Snippets for Options pattern * lint/bug fixes. --- .../apache/beam/examples/snippets/Snippets.java | 48 +++++++++++++++++++++ .../apache_beam/examples/snippets/snippets.py | 49 ++++++++++++++++++++++ 2 files changed, 97 insertions(+) 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 8216bba..4d01feb 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 @@ -42,6 +42,11 @@ import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition; import org.apache.beam.sdk.io.gcp.bigquery.DynamicDestinations; import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord; import org.apache.beam.sdk.io.gcp.bigquery.TableDestination; +import org.apache.beam.sdk.options.Default; +import org.apache.beam.sdk.options.Description; +import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.options.PipelineOptionsFactory; +import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.transforms.Create; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.MapElements; @@ -605,4 +610,47 @@ public class Snippets { // [END SideInputPatternSlowUpdateGlobalWindowSnip1] + // [START AccessingValueProviderInfoAfterRunSnip1] + + /** Sample of PipelineOptions with a ValueProvider option argument. */ + public interface MyOptions extends PipelineOptions { + @Description("My option") + @Default.String("Hello world!") + ValueProvider<String> getStringValue(); + + void setStringValue(ValueProvider<String> value); + } + + public static void accessingValueProviderInfoAfterRunSnip1(String[] args) { + + MyOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(MyOptions.class); + + // Create pipeline. + Pipeline p = Pipeline.create(options); + + // Add a branch for logging the ValueProvider value. + p.apply(Create.of(1)) + .apply( + ParDo.of( + new DoFn<Integer, Integer>() { + + // Define the DoFn that logs the ValueProvider value. + @ProcessElement + public void process(ProcessContext c) { + + MyOptions ops = c.getPipelineOptions().as(MyOptions.class); + // This example logs the ValueProvider value, but you could store it by + // pushing it to an external database. + + LOG.info("Option StringValue was {}", ops.getStringValue()); + } + })); + + // The main pipeline. + p.apply(Create.of(1, 2, 3, 4)).apply(Sum.integersGlobally()); + + p.run(); + } + + // [END AccessingValueProviderInfoAfterRunSnip1] } diff --git a/sdks/python/apache_beam/examples/snippets/snippets.py b/sdks/python/apache_beam/examples/snippets/snippets.py index 5e7f9bf..7111c99 100644 --- a/sdks/python/apache_beam/examples/snippets/snippets.py +++ b/sdks/python/apache_beam/examples/snippets/snippets.py @@ -1352,3 +1352,52 @@ def file_process_pattern_access_metadata(): | beam.Map(lambda x: (x.metadata.path, x.read_utf8()))) # [END FileProcessPatternAccessMetadataSnip1] + + +def accessing_valueprovider_info_after_run(): + # [START AccessingValueProviderInfoAfterRunSnip1] + import logging + + import apache_beam as beam + from apache_beam.options.pipeline_options import PipelineOptions + from apache_beam.utils.value_provider import RuntimeValueProvider + from apache_beam.io import WriteToText + + class MyOptions(PipelineOptions): + @classmethod + def _add_argparse_args(cls, parser): + parser.add_value_provider_argument('--string_value', type=str) + + class LogValueProvidersFn(beam.DoFn): + def __init__(self, string_vp): + self.string_vp = string_vp + + # Define the DoFn that logs the ValueProvider value. + # The DoFn is called when creating the pipeline branch. + # This example logs the ValueProvider value, but + # you could store it by pushing it to an external database. + def process(self, an_int): + logging.info('The string_value is %s' % self.string_vp.get()) + # Another option (where you don't need to pass the value at all) is: + logging.info('The string value is %s' % + RuntimeValueProvider.get_value('string_value', str, '')) + + pipeline_options = PipelineOptions() + # Create pipeline. + p = beam.Pipeline(options=pipeline_options) + + my_options = pipeline_options.view_as(MyOptions) + # Add a branch for logging the ValueProvider value. + _ = (p + | beam.Create([None]) + | 'LogValueProvs' >> beam.ParDo( + LogValueProvidersFn(my_options.string_value))) + + # The main pipeline. + result_pc = (p + | "main_pc" >> beam.Create([1, 2, 3]) + | beam.combiners.Sum.Globally()) + + p.run().wait_until_finish() + + # [END AccessingValueProviderInfoAfterRunSnip1]