Skips a flakky airflow test on macos.

Recent versions of Tensorflow and Airflow makes test fails with segmentation faults.
Airflow doesn't support macos as an execution environment.[1]

[1] apache/airflow#11965 (comment)

PiperOrigin-RevId: 364952307

Author: jiyongjung

tfx/dependencies.py

@@ -164,10 +164,7 @@ def make_extra_packages_examples():
       'tensorflow-ranking>=0.3.3,<0.4',
       'struct2tensor>=0.28,<0.29',
       # Required for tfx/examples/penguin/experimental
-      # LINT.IfChange
-      'scikit-learn>=0.23,<0.24',
-      # LINT.ThenChange(
-      #     examples/penguin/experimental/penguin_pipeline_sklearn_gcp.py)
+      'scikit-learn>=0.24,<0.25',
   ]
 
 

tfx/dsl/component/experimental/executor_specs.py

@@ -74,18 +74,20 @@ class MyTrainerSpec(types.ComponentSpec):
     command: Container entrypoint command-line. Not executed within a shell.
       The command-line can use placeholder objects that will be replaced at
       the compilation time. Note: Jinja templates are not supported.
-    args: Container entrypoint command-args.
   """
 
+  # The "command" parameter holds the name of the program and its arguments.
+  # The "command" parameter is required to enable instrumentation.
+  # The command-line is often split into command+args, but here "args" would be
+  # redundant since all items can just be added to "command".
+
   def __init__(
       self,
       image: Text,
-      command: Optional[List[placeholders.CommandlineArgumentType]] = None,
-      args: Optional[List[placeholders.CommandlineArgumentType]] = None,
+      command: List[placeholders.CommandlineArgumentType],
   ):
     self.image = image
-    self.command = command or []
-    self.args = args or []
+    self.command = command
     super(TemplatedExecutorContainerSpec, self).__init__()
 
   def __eq__(self, other) -> bool:
@@ -96,27 +98,27 @@ def __ne__(self, other) -> bool:
     return not self.__eq__(other)
 
   def _recursively_encode(
-      self, ph: placeholders.CommandlineArgumentType
+      self, command: placeholders.CommandlineArgumentType
   ) -> Union[str, placeholder.Placeholder]:
-    if isinstance(ph, str):
-      return ph
-    elif isinstance(ph, placeholders.InputValuePlaceholder):
-      return placeholder.input(ph.input_name)[0]
-    elif isinstance(ph, placeholders.InputUriPlaceholder):
-      return placeholder.input(ph.input_name)[0].uri
-    elif isinstance(ph, placeholders.OutputUriPlaceholder):
-      return placeholder.output(ph.output_name)[0].uri
-    elif isinstance(ph, placeholders.ConcatPlaceholder):
+    if isinstance(command, str):
+      return command
+    elif isinstance(command, placeholders.InputValuePlaceholder):
+      return placeholder.input(command.input_name)[0]
+    elif isinstance(command, placeholders.InputUriPlaceholder):
+      return placeholder.input(command.input_name)[0].uri
+    elif isinstance(command, placeholders.OutputUriPlaceholder):
+      return placeholder.output(command.output_name)[0].uri
+    elif isinstance(command, placeholders.ConcatPlaceholder):
       # operator.add wil use the overloaded __add__ operator for Placeholder
       # instances.
       return functools.reduce(
           operator.add,
-          [self._recursively_encode(item) for item in ph.items])
+          [self._recursively_encode(item) for item in command.items])
     else:
       raise TypeError(
-          ('Unsupported type of placeholder arguments: "{}".'
+          ('Unsupported type of command-line arguments: "{}".'
            ' Supported types are {}.')
-          .format(type(ph), str(placeholders.CommandlineArgumentType)))
+          .format(type(command), str(placeholders.CommandlineArgumentType)))
 
   def encode(
       self,
@@ -142,14 +144,4 @@ def encode(
         cmd.CopyFrom(expression)
       else:
         cmd.CopyFrom(self._recursively_encode(command).encode())
-
-    for arg in self.args:
-      cmd = result.args.add()
-      str_or_placeholder = self._recursively_encode(arg)
-      if isinstance(str_or_placeholder, str):
-        expression = placeholder_pb2.PlaceholderExpression()
-        expression.value.string_value = str_or_placeholder
-        cmd.CopyFrom(expression)
-      else:
-        cmd.CopyFrom(self._recursively_encode(arg).encode())
     return result

tfx/dsl/component/experimental/executor_specs_test.py

@@ -166,9 +166,6 @@ def testEncodeTemplatedExecutorContainerSpec_withConcatAllText(self):
         image='image',
         command=[
             self._text_concat_placeholder
-        ],
-        args=[
-            self._text_concat_placeholder
         ])
     encode_result = specs.encode()
     self.assertProtoEquals("""
@@ -177,11 +174,6 @@ def testEncodeTemplatedExecutorContainerSpec_withConcatAllText(self):
         value {
           string_value: "texttext1text2"
         }
-      }
-      args {
-        value {
-          string_value: "texttext1text2"
-        }
       }""", encode_result)
 
 

tfx/examples/chicago_taxi_pipeline/taxi_pipeline_simple_airflow_e2e_test.py

@@ -14,9 +14,11 @@
 """End to end test for tfx.orchestration.airflow."""
 
 import os
+import platform
 import subprocess
 import time
 from typing import Sequence, Set, Text
+import unittest
 
 import absl
 import tensorflow as tf
@@ -54,6 +56,11 @@ def __exit__(self, exception_type, exception_value, traceback):  # pylint: disab
 _PENDING_TASK_STATES = set(['queued', 'scheduled', 'running', 'none'])
 
 
+@unittest.skipIf(
+    platform.system() == 'Darwin',
+    'Airflow is not compatible with TF in some environments on macos and '
+    'Airflow Executor is not supported on macos. See b/178137745.'
+)
 class AirflowEndToEndTest(test_case_utils.TfxTest):
   """An end to end test using fully orchestrated Airflow."""
 

tfx/examples/penguin/experimental/Dockerfile

@@ -13,5 +13,8 @@
 # limitations under the License.
 
 ARG TFX_VERSION
-FROM tensorflow/tfx:$TFX_VERSION
-RUN pip install -U tfx[examples]
+FROM gcr.io/tfx-oss-public/tfx:$TFX_VERSION
+
+RUN pip install \
+    scikit-learn==0.20.4 \
+    numpy==1.16.5

tfx/examples/penguin/experimental/README.md

@@ -45,9 +45,10 @@ image and upload it to Google Container Registry (GCR).
 <pre class="devsite-terminal devsite-click-to-copy">
 gcloud auth configure-docker
 docker build \
-  --tag gcr.io/[PROJECT-ID]/tfx-example-sklearn
+  --tag tfx-example-sklearn \
   --build-arg TFX_VERSION=$(python -c 'import tfx; print(tfx.__version__)') \
   .
+docker tag tfx-example-sklearn gcr.io/[PROJECT-ID]/tfx-example-sklearn
 docker push gcr.io/[PROJECT-ID]/tfx-example-sklearn
 </pre>
 

tfx/examples/penguin/experimental/penguin_pipeline_sklearn_gcp.py

@@ -86,11 +86,9 @@
     # Note that serving currently only supports a single region:
     # https://cloud.google.com/ml-engine/reference/rest/v1/projects.models#Model
     'regions': [_gcp_region],
-    # TODO(b/176256164): Update to runtime version 2.4 once that is available
-    # to align with the version of TF supported by TFX.
-    # LINT.IfChange
-    'runtime_version': '2.3',
-    # LINT.ThenChange(../../../dependencies.py)
+    # TODO(b/157646655): Update the version once sklearn support is added back
+    # to CAIP in the next runtime release.
+    'runtime_version': '1.15',
 }
 
 # This example assumes that Penguin data is stored in ~/penguin/data and the
@@ -160,6 +158,7 @@ def _create_pipeline(pipeline_name: Text, pipeline_root: Text, data_root: Text,
   # Uses user-provided Python function that trains a model using TF-Learn.
   # Num_steps is not provided during evaluation because the scikit-learn model
   # loads and evaluates the entire test set at once.
+  # TODO(b/159470716): Make schema optional in Trainer.
   trainer = Trainer(
       module_file=trainer_module_file,
       custom_executor_spec=executor_spec.ExecutorClassSpec(

tfx/examples/penguin/experimental/penguin_pipeline_sklearn_local.py

@@ -106,6 +106,7 @@ def _create_pipeline(pipeline_name: Text, pipeline_root: Text, data_root: Text,
   # Uses user-provided Python function that trains a model using TF-Learn.
   # Num_steps is not provided during evaluation because the scikit-learn model
   # loads and evaluates the entire test set at once.
+  # TODO(b/159470716): Make schema optional in Trainer.
   trainer = Trainer(
       module_file=trainer_module_file,
       custom_executor_spec=executor_spec.ExecutorClassSpec(GenericExecutor),

tfx/extensions/google_cloud_ai_platform/runner.py

@@ -377,7 +377,7 @@ def deploy_model_for_aip_prediction(api: discovery.Resource,
     if e.resp.status == 409:
       logging.warn('Model version %s already exists', model_version)
     else:
-      raise RuntimeError('Creating model version to AI Platform failed: {}'
+      raise RuntimeError('Creating model verseion to AI Platform failed: {}'
                          .format(e))
 
   if set_default_version:

tfx/orchestration/kubeflow/container_entrypoint.py

@@ -310,10 +310,6 @@ def main():
   parser.add_argument('--serialized_component', type=str, required=True)
   parser.add_argument('--component_config', type=str, required=True)
 
-  # TODO(b/182220464): Uses following flags and change to required.
-  parser.add_argument('--tfx_ir', type=str, required=False)
-  parser.add_argument('--node_id', type=str, required=False)
-
   args = parser.parse_args()
 
   component = json_utils.loads(args.serialized_component)

tfx/tools/cli/handler/base_handler.py

@@ -191,11 +191,11 @@ def _subprocess_call(self,
   def _format_table(self, header: Collection[Any],
                     data: Collection[Collection[Any]]):
 
-    def _format_as_strings(items):
-      return [f' {item} ' for item in items]
+    def _force_strings(items):
+      return [str(item) for item in items]
 
-    header = _format_as_strings(header)
-    data = [_format_as_strings(row) for row in data]
+    header = _force_strings(header)
+    data = [_force_strings(row) for row in data]
 
     max_widths = [len(s) for s in header]
     for row in data:

tfx/tools/cli/handler/base_handler_test.py

@@ -207,13 +207,13 @@ def testFormatTable(self):
     handler = FakeHandler(flags_dict)
     self.assertEqual(
         textwrap.dedent("""\
-      +=====+=====+=======+
-      | abc | d   | False |
-      +=====+=====+=======+
-      | 1   | 234 | None  |
-      +-----+-----+-------+
-      | xxx |     | []    |
-      +=====+=====+=======+
+      +===+===+=====+
+      |abc|d  |False|
+      +===+===+=====+
+      |1  |234|None |
+      +---+---+-----+
+      |xxx|   |[]   |
+      +===+===+=====+
       """),
         handler._format_table(('abc', 'd', False),
                               [[1, '234', None], ['xxx', '', []]]))