Logging pipeline messages snippet

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=117286960

Author: aaltay

google/cloud/dataflow/examples/snippets/snippets.py

@@ -285,7 +285,7 @@ def _add_argparse_args(cls, parser):
 
 
 def pipeline_options_command_line(argv):
-  """"Creating a Pipeline by passing a list of arguments.
+  """Creating a Pipeline by passing a list of arguments.
 
   URL: https://cloud.google.com/dataflow/pipelines/specifying-exec-params
   """
@@ -308,6 +308,43 @@ def pipeline_options_command_line(argv):
   p.run()
 
 
+def pipeline_logging(lines, output):
+  """Logging Pipeline Messages.
+
+  URL: https://cloud.google.com/dataflow/pipelines/logging
+  """
+
+  import re
+  import google.cloud.dataflow as df
+  from google.cloud.dataflow.utils.options import PipelineOptions
+
+  # [START pipeline_logging]
+  # import Python logging module.
+  import logging
+
+  class ExtractWordsFn(df.DoFn):
+
+    def process(self, context):
+      words = re.findall(r'[A-Za-z\']+', context.element)
+      for word in words:
+        yield word
+
+        if word.lower() == 'love':
+          # Log using the root logger at info or higher levels
+          logging.info('Found : %s', word.lower())
+
+  # Remaining WordCount example code ...
+  # [END pipeline_logging]
+
+  p = df.Pipeline(options=PipelineOptions())
+  (p
+   | df.Create(lines)
+   | df.ParDo('ExtractWords', ExtractWordsFn())
+   | df.io.Write('WriteToText', df.io.TextFileSink(output)))
+
+  p.run()
+
+
 def model_textio(renames):
   """Using a Read and Write transform to read/write text files.
 

google/cloud/dataflow/examples/snippets/snippets_test.py

@@ -392,6 +392,16 @@ def test_pipeline_options_remote(self):
   def test_pipeline_options_command_line(self):
     self._run_test_pipeline_for_options(snippets.pipeline_options_command_line)
 
+  def test_pipeline_logging(self):
+    result_path = self.create_temp_file()
+    lines = ['we found love right where we are',
+             'we found love right from the start',
+             'we found love in a hopeless place']
+    snippets.pipeline_logging(lines, result_path)
+    self.assertEqual(
+        sorted(' '.join(lines).split(' ')),
+        self.get_output(result_path))
+
   def test_model_composite_transform_example(self):
     contents = ['aa bb cc', 'bb cc', 'cc']
     result_path = self.create_temp_file()