Update calls to a since-renamed beam API. (#694)

* Update calls to a since-renamed beam API.

The `beam.pvalue.SideOutputValue` method was renamed to
`beam.pvalue.TaggedOutput` in release
[2.0.0](https://cloud.google.com/dataflow/release-notes/release-notes-python#200_may_31_2017)
of the Cloud Dataflow SDK.

This change updates calls to that method to instead use the new name.

* Upgrade the TOX environment

* Report prediction errors

Author: ojarjur

solutionbox/ml_workbench/tensorflow/transform.py

@@ -349,8 +349,7 @@ def process(self, element):
         yield transformed_features
 
     except Exception as e:  # pylint: disable=broad-except
-      yield beam.pvalue.SideOutputValue('errors',
-                                        (str(e), element))
+      yield beam.pvalue.TaggedOutput('errors', (str(e), element))
 
 
 def decode_csv(csv_string, column_names):

solutionbox/ml_workbench/xgboost/transform.py

@@ -349,8 +349,7 @@ def process(self, element):
         yield transformed_features
 
     except Exception as e:  # pylint: disable=broad-except
-      yield beam.pvalue.SideOutputValue('errors',
-                                        (str(e), element))
+      yield beam.pvalue.TaggedOutput('errors', (str(e), element))
 
 
 def decode_csv(csv_string, column_names):

solutionbox/structured_data/mltoolbox/_structured_data/prediction/predict.py

@@ -215,8 +215,7 @@ def process(self, element):
         yield predictions
 
     except Exception as e:  # pylint: disable=broad-except
-      yield beam.pvalue.SideOutputValue('errors',
-                                        (str(e), element))
+      yield beam.pvalue.TaggedOutput('errors', (str(e), element))
 
 
 class RawJsonCoder(beam.coders.Coder):

solutionbox/structured_data/test_mltoolbox/test_datalab_e2e.py

@@ -152,7 +152,9 @@ def _run_batch_prediction(self, output_dir, use_target):
     # check errors file is empty
     errors = file_io.get_matching_files(os.path.join(output_dir, 'errors*'))
     self.assertEqual(len(errors), 1)
-    self.assertEqual(os.path.getsize(errors[0]), 0)
+    if os.path.getsize(errors[0]):
+      with open(errors[0]) as errors_file:
+        self.fail(msg=errors_file.read())
 
     # check predictions files are not empty
     predictions = file_io.get_matching_files(os.path.join(output_dir,

tox.ini

@@ -27,9 +27,9 @@ commands =
   python ./legacy_tests/main.py
 
 [testenv:py27]
-# google-cloud-dataflow only supports python2.7, so we add that here.
+# apache-beam only supports python2.7, so we add that here.
 deps = {[testenv]deps}
-       google-cloud-dataflow==2.5.0
+       apache-beam==2.5.0
 
 [testenv:flake8]
 commands = flake8 --exclude=.tox,.git,./*.egg,build,.cache,env,__pycache__,docs