Merge branch 'tf-transformer-part1' into api-tf-transformer

Author: phi-dbq

python/sparkdl/param/converters.py

@@ -13,6 +13,8 @@
 # limitations under the License.
 #
 
+import six
+
 import tensorflow as tf
 
 from pyspark.ml.param import TypeConverters
@@ -21,6 +23,37 @@
 from sparkdl.graph.input import TFInputGraph
 import sparkdl.utils.keras_model as kmutil
 
+__all__ = ['SparkDLTypeConverters']
+
+def _try_convert_tf_tensor_mapping(value, is_key_tf_tensor=True):
+    if isinstance(value, dict):
+        strs_pair_seq = []
+        for k, v in value.items():
+            try:
+                if is_key_tf_tensor:
+                    _pair = (tfx.as_tensor_name(k), v)
+                else:
+                    _pair = (k, tfx.as_tensor_name(v))
+            except:
+                err_msg = "Can NOT convert {} (type {}) to tf.Tensor name"
+                _not_tf_op = k if is_key_tf_tensor else v
+                raise TypeError(err_msg.format(_not_tf_op, type(_not_tf_op)))
+
+            str_val = v if is_key_tf_tensor else k
+            if not isinstance(str_val, six.string_types):
+                err_msg = 'expect string type for {}, but got {}'
+                raise TypeError(err_msg.format(str_val, type(str_val)))
+
+            strs_pair_seq.append(_pair)
+
+        return sorted(strs_pair_seq)
+
+    if is_key_tf_tensor:
+        raise TypeError("Could not convert %s to tf.Tensor name to str mapping" % type(value))
+    else:
+        raise TypeError("Could not convert %s to str to tf.Tensor name mapping" % type(value))
+
+
 class SparkDLTypeConverters(object):
     @staticmethod
     def toTFGraph(value):
@@ -37,18 +70,12 @@ def toTFInputGraph(value):
             raise TypeError("Could not convert %s to TFInputGraph" % type(value))
 
     @staticmethod
-    def asColumnToTensorMap(value):
-        if isinstance(value, dict):
-            strs_pair_seq = [(k, tfx.as_op_name(v)) for k, v in value.items()]
-            return sorted(strs_pair_seq)
-        raise TypeError("Could not convert %s to TensorFlow Tensor" % type(value))
+    def asColumnToTensorNameMap(value):
+        return _try_convert_tf_tensor_mapping(value, is_key_tf_tensor=False)
 
     @staticmethod
-    def asTensorToColumnMap(value):
-        if isinstance(value, dict):
-            strs_pair_seq = [(tfx.as_op_name(k), v) for k, v in value.items()]
-            return sorted(strs_pair_seq)
-        raise TypeError("Could not convert %s to TensorFlow Tensor" % type(value))
+    def asTensorNameToColumnMap(value):
+        return _try_convert_tf_tensor_mapping(value, is_key_tf_tensor=True)
 
     @staticmethod
     def toTFHParams(value):

python/sparkdl/param/shared_params.py

@@ -22,17 +22,10 @@
 from functools import wraps
 import six
 
-import keras
-import tensorflow as tf
-
 from pyspark.ml.param import Param, Params, TypeConverters
 
-from sparkdl.graph.builder import GraphFunction, IsolatedSession
-import sparkdl.graph.utils as tfx
 from sparkdl.graph.input import TFInputGraph
 from sparkdl.param.converters import SparkDLTypeConverters
-import sparkdl.utils.keras_model as kmutil
-
 
 ########################################################
 # Copied from PySpark for backward compatibility.
@@ -204,11 +197,10 @@ class HasOutputMapping(Params):
     """
     Mixin for param outputMapping: ordered list of ('outputTensorOpName', 'outputColName') pairs
     """
-    outputMapping = Param(
-        Params._dummy(),
-        "outputMapping",
-        "Mapping output :class:`tf.Operation` names to DataFrame column names",
-        typeConverter=SparkDLTypeConverters.asTensorToColumnMap)
+    outputMapping = Param(Params._dummy(),
+                          "outputMapping",
+                          "Mapping output :class:`tf.Tensor` names to DataFrame column names",
+                          typeConverter=SparkDLTypeConverters.asTensorNameToColumnMap)
 
     def setOutputMapping(self, value):
         # NOTE(phi-dbq): due to the nature of TensorFlow import modes, we can only derive the
@@ -225,11 +217,10 @@ class HasInputMapping(Params):
     """
     Mixin for param inputMapping: ordered list of ('inputColName', 'inputTensorOpName') pairs
     """
-    inputMapping = Param(
-        Params._dummy(),
-        "inputMapping",
-        "Mapping input DataFrame column names to :class:`tf.Operation` names",
-        typeConverter=SparkDLTypeConverters.asColumnToTensorMap)
+    inputMapping = Param(Params._dummy(),
+                         "inputMapping",
+                         "Mapping input DataFrame column names to :class:`tf.Tensor` names",
+                         typeConverter=SparkDLTypeConverters.asColumnToTensorNameMap)
 
     def setInputMapping(self, value):
         # NOTE(phi-dbq): due to the nature of TensorFlow import modes, we can only derive the
@@ -246,11 +237,10 @@ class HasTFInputGraph(Params):
     """
     Mixin for param tfInputGraph: a serializable object derived from a TensorFlow computation graph.
     """
-    tfInputGraph = Param(
-        Params._dummy(),
-        "tfInputGraph",
-        "A serializable object derived from a TensorFlow computation graph",
-        typeConverter=SparkDLTypeConverters.toTFInputGraph)
+    tfInputGraph = Param(Params._dummy(),
+                         "tfInputGraph",
+                         "A serializable object derived from a TensorFlow computation graph",
+                         typeConverter=SparkDLTypeConverters.toTFInputGraph)
 
     def __init__(self):
         super(HasTFInputGraph, self).__init__()
@@ -271,11 +261,10 @@ class HasTFHParams(Params):
     """
     Mixin for TensorFlow model hyper-parameters
     """
-    tfHParams = Param(
-        Params._dummy(),
-        "hparams",
-        "instance of :class:`tf.contrib.training.HParams`, a key-value map-like object",
-        typeConverter=SparkDLTypeConverters.toTFHParams)
+    tfHParams = Param(Params._dummy(),
+                      "hparams",
+                      "instance of :class:`tf.contrib.training.HParams`, a key-value map-like object",
+                      typeConverter=SparkDLTypeConverters.toTFHParams)
 
     def setTFHParams(self, value):
         return self._set(tfHParam=value)

python/tests/param/__init__.py

@@ -0,0 +1,15 @@
+#
+# Copyright 2017 Databricks, Inc.
+#
+# 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.
+#

python/tests/param/params_test.py

@@ -0,0 +1,69 @@
+# Copyright 2017 Databricks, Inc.
+#
+# 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.
+#
+import sys
+
+if sys.version_info[:2] <= (2, 6):
+    try:
+        import unittest2 as unittest
+    except ImportError:
+        sys.stderr.write('Please install unittest2 to test with Python 2.6 or earlier')
+        sys.exit(1)
+else:
+    import unittest
+
+from sparkdl.param.converters import SparkDLTypeConverters as conv
+
+class ParamsConverterTest(unittest.TestCase):
+    # pylint: disable=protected-access
+
+    def test_tf_input_mapping_converter(self):
+        valid_tnsr_input = {'colA': 'tnsrOpA:0',
+                            'colB': 'tnsrOpB:0'}
+        valid_op_input = {'colA': 'tnsrOpA',
+                          'colB': 'tnsrOpB'}
+        valid_input_mapping_result = [('colA', 'tnsrOpA:0'),
+                                      ('colB', 'tnsrOpB:0')]
+
+        for valid_input_mapping in [valid_op_input, valid_tnsr_input]:
+            res = conv.asColumnToTensorNameMap(valid_input_mapping)
+            self.assertEqual(valid_input_mapping_result, res)
+
+    def test_tf_output_mapping_converter(self):
+        valid_tnsr_output = {'tnsrOpA:0': 'colA',
+                             'tnsrOpB:0': 'colB'}
+        valid_op_output = {'tnsrOpA': 'colA',
+                           'tnsrOpB': 'colB'}
+        valid_output_mapping_result = [('tnsrOpA:0', 'colA'),
+                                       ('tnsrOpB:0', 'colB')]
+
+        for valid_output_mapping in [valid_tnsr_output, valid_op_output]:
+            res = conv.asTensorNameToColumnMap(valid_output_mapping)
+            self.assertEqual(valid_output_mapping_result, res)
+
+
+    def test_invalid_input_mapping(self):
+        for invalid in [['a1', 'b2'], ('c3', 'd4'), [('a', 1), ('b', 2)]]:
+            with self.assertRaises(TypeError):
+                conv.asColumnToTensorNameMap(invalid)
+                conv.asTensorNameToColumnMap(invalid)
+
+        with self.assertRaises(TypeError):
+            # Wrong value type: must be string
+            conv.asTensorNameToColumnMap({1: 'a', 2.0: 'b'})
+            conv.asColumnToTensorNameMap({'a': 1, 'b': 2.0})
+
+            # Wrong containter type: only accept dict
+            conv.asColumnToTensorNameMap([('colA', 'tnsrA:0'), ('colB', 'tnsrB:0')])
+            conv.asTensorNameToColumnMap([('tnsrA:0', 'colA'), ('tnsrB:0', 'colB')])