Replace sparkdl's ImageSchema with Spark2.3's version (#85)

Use Spark 2.3's ImageSchema as image interface.

The biggest change is using opposite ordering of color channels - BGR instead of RGB, requires extra reordering in various places. The change affects mostly the transformers and the udf creation functionality. Some noteworthy decisions:
- For DeepImageFeaturizer & DeepImagePredictor, we preserved ability to read and resize images in python using PIL to match Keras. Those image read & resize utilities are not recommended for external use as it's likely to cause confusion.
- For KerasImageFileTransformer and the keras udf creator, we assume the preprocessing function & model inputs work on RGB images since Keras works with RGB images.
- For TFImageTransformer, we added a param to specify the channel ordering expected by the tf.graph’s input layer. Having this param explicitly raises awareness that you could be doing the wrong thing, and makes the code easier to reason about.

Also needed a few tweaks to run with spark 2.3 - notably UDFs are now referenced by SQL identifier and can not have dash as part of the name.

[TODO]
- In order to run on spark < 2.3, the image schema files have been copied here and need to be removed in the future once Spark 2.3 is released.
- During this work we discovered that ImageSchema-related utilities in Spark 2.3 should support float32 types and a bit more info about the modes. Once that is done we can remove some code from this PR and use the Spark 2.3 version instead.

Author: tomasatdatabricks

README.md

@@ -80,14 +80,23 @@ To try running the examples below, check out the Databricks notebook [DeepLearni
 
 ### Working with images in Spark
 
-The first step to applying deep learning on images is the ability to load the images. Deep Learning Pipelines includes utility functions that can load millions of images into a Spark DataFrame and decode them automatically in a distributed fashion, allowing manipulation at scale.
+The first step to applying deep learning on images is the ability to load the images. Spark and Deep Learning Pipelines include utility functions that can load millions of images into a Spark DataFrame and decode them automatically in a distributed fashion, allowing manipulation at scale.
+
+Using Spark's ImageSchema
+
+```python
+from sparkdl.image.image import ImageSchema
+image_df = ImageSchema.readImages("/data/myimages")
+```
+
+or if custom image library is needed:
 
 ```python
-from sparkdl import readImages
-image_df = readImages("/data/myimages")
+from sparkdl.image import imageIO as imageIO
+image_df = imageIO.readImagesWithCustomFn("/data/myimages",decode_f=<your image library, see imageIO.PIL_decode>)
 ```
 
-The resulting DataFrame contains a string column named "filePath" containing the path to each image file, and a image struct ("`SpImage`") column named "image" containing the decoded image data.
+The resulting DataFrame contains a string column named "image" containing an image struct with schema == ImageSchema.
 
 ```python
 image_df.show()
@@ -109,7 +118,7 @@ featurizer = DeepImageFeaturizer(inputCol="image", outputCol="features", modelNa
 lr = LogisticRegression(maxIter=20, regParam=0.05, elasticNetParam=0.3, labelCol="label")
 p = Pipeline(stages=[featurizer, lr])
 
-model = p.fit(train_images_df)    # train_images_df is a dataset of images (SpImage) and labels
+model = p.fit(train_images_df)    # train_images_df is a dataset of images and labels
 
 # Inspect training error
 df = model.transform(train_images_df.limit(10)).select("image", "probability",  "uri", "label")
@@ -127,11 +136,13 @@ Spark DataFrames are a natural construct for applying deep learning models to a
     There are many well-known deep learning models for images. If the task at hand is very similar to what the models provide (e.g. object recognition with ImageNet classes), or for pure exploration, one can use the Transformer `DeepImagePredictor` by simply specifying the model name.
 
     ```python
-    from sparkdl import readImages, DeepImagePredictor
+    from sparkdl.image.image import ImageSchema
+
+    from sparkdl import DeepImagePredictor
 
     predictor = DeepImagePredictor(inputCol="image", outputCol="predicted_labels",
                                    modelName="InceptionV3", decodePredictions=True, topK=10)
-    image_df = readImages("/data/myimages")
+    image_df = ImageSchema.readImages("/data/myimages")
     predictions_df = predictor.transform(image_df)
     ```
 
@@ -140,7 +151,8 @@ Spark DataFrames are a natural construct for applying deep learning models to a
     Deep Learning Pipelines provides a Transformer that will apply the given TensorFlow Graph to a DataFrame containing a column of images (e.g. loaded using the utilities described in the previous section). Here is a very simple example of how a TensorFlow Graph can be used with the Transformer. In practice, the TensorFlow Graph will likely be restored from files before calling `TFImageTransformer`.
 
     ```python
-    from sparkdl import readImages, TFImageTransformer
+    from sparkdl.image.image import ImageSchema
+    from sparkdl import TFImageTransformer
     import sparkdl.graph.utils as tfx
     from sparkdl.transformers import utils
     import tensorflow as tf
@@ -155,7 +167,7 @@ Spark DataFrames are a natural construct for applying deep learning models to a
     transformer = TFImageTransformer(inputCol="image", outputCol="predictions", graph=frozen_graph,
                                      inputTensor=image_arr, outputTensor=resized_images,
                                      outputMode="image")
-    image_df = readImages("/data/myimages")
+    image_df = ImageSchema.readImages("/data/myimages")
     processed_image_df = transformer.transform(image_df)
     ```
 

build.sbt

@@ -35,7 +35,7 @@ sparkComponents ++= Seq("mllib-local", "mllib", "sql")
 // add any Spark Package dependencies using spDependencies.
 // e.g. spDependencies += "databricks/spark-avro:0.1"
 spDependencies += s"databricks/tensorframes:0.2.9-s_${scalaMajorVersion}"
-spDependencies += "Microsoft/spark-images:0.1"
+
 
 // These versions are ancient, but they cross-compile around scala 2.10 and 2.11.
 // Update them when dropping support for scala 2.10

project/plugins.sbt

@@ -1,8 +1,5 @@
 // You may use this file to add plugin dependencies for sbt.
 resolvers += "Spark Packages repo" at "https://dl.bintray.com/spark-packages/maven/"
-
 addSbtPlugin("org.spark-packages" %% "sbt-spark-package" % "0.2.5")
-
 // scalacOptions in (Compile,doc) := Seq("-groups", "-implicits")
-
 addSbtPlugin("org.scoverage" % "sbt-scoverage" % "1.5.0")

python/sparkdl/__init__.py

@@ -12,9 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-
 from .graph.input import TFInputGraph
-from .image.imageIO import imageSchema, imageType, readImages
 from .transformers.keras_image import KerasImageFileTransformer
 from .transformers.named_image import DeepImagePredictor, DeepImageFeaturizer
 from .transformers.tf_image import TFImageTransformer

python/sparkdl/estimators/keras_image_file_estimator.py

@@ -36,6 +36,7 @@
 
 logger = logging.getLogger('sparkdl')
 
+
 class KerasImageFileEstimator(Estimator, HasInputCol, HasInputImageNodeName,
                               HasOutputCol, HasOutputNodeName, HasLabelCol,
                               HasKerasModel, HasKerasOptimizer, HasKerasLoss,

python/sparkdl/graph/builder.py

@@ -27,6 +27,7 @@
 
 logger = logging.getLogger('sparkdl')
 
+
 class IsolatedSession(object):
     """
     Provide an isolated session to work with mixed Keras and TensorFlow
@@ -43,6 +44,7 @@ class IsolatedSession(object):
                         In this case, all Keras models loaded in this session will be accessible
                         as a subgraph of of `graph`
     """
+
     def __init__(self, graph=None, using_keras=False):
         self.graph = graph or tf.Graph()
         self.sess = tf.Session(graph=self.graph)
@@ -166,7 +168,7 @@ def _fromKerasModelFile(cls, file_path):
             'Keras model must be specified as HDF5 file'
 
         with IsolatedSession(using_keras=True) as issn:
-            K.set_learning_phase(0) # Testing phase
+            K.set_learning_phase(0)  # Testing phase
             model = load_model(file_path)
             gfn = issn.asGraphFunction(model.inputs, model.outputs)
 
@@ -223,7 +225,8 @@ def fromList(cls, functions):
             # We currently only support single input/output for intermediary stages
             # The functions could still take multi-dimensional tensor, but only one
             if len(gfn_out.input_names) != 1:
-                raise NotImplementedError("Only support single input/output for intermediary layers")
+                raise NotImplementedError(
+                    "Only support single input/output for intermediary layers")
 
         # Acquire initial placeholders' properties
         # We want the input names of the merged function are not under scoped

python/sparkdl/graph/input.py

@@ -23,6 +23,7 @@
 
 # pylint: disable=invalid-name,wrong-spelling-in-comment,wrong-spelling-in-docstring
 
+
 class TFInputGraph(object):
     """
     An opaque object containing TensorFlow graph.
@@ -84,7 +85,6 @@ class TFInputGraph(object):
                                               Please see the example above.
     """
 
-
     def __init__(self, graph_def, input_tensor_name_from_signature,
                  output_tensor_name_from_signature):
         self.graph_def = graph_def
@@ -281,6 +281,7 @@ def _from_checkpoint_impl(checkpoint_dir, signature_def_key, feed_names, fetch_n
             return _build_with_feeds_fetches(sess=sess, graph=graph, feed_names=feed_names,
                                              fetch_names=fetch_names)
 
+
 def _from_saved_model_impl(saved_model_dir, tag_set, signature_def_key, feed_names, fetch_names):
     """
     Construct a TFInputGraph from a SavedModel.

python/sparkdl/graph/pieces.py

@@ -18,7 +18,7 @@
 import tensorflow as tf
 
 from sparkdl.graph.builder import IsolatedSession
-from sparkdl.image.imageIO import SparkMode
+from sparkdl.image import imageIO
 
 logger = logging.getLogger('sparkdl')
 
@@ -29,7 +29,8 @@
       Deserializing ProtocolBuffer bytes is in general faster than directly loading Keras models.
 """
 
-def buildSpImageConverter(img_dtype):
+
+def buildSpImageConverter(channelOrder, img_dtype):
     """
     Convert a imageIO byte encoded image into a image tensor suitable as input to ConvNets
     The name of the input must be a subset of those specified in `image.imageIO.imageSchema`.
@@ -48,23 +49,25 @@ def buildSpImageConverter(img_dtype):
         # This is the default behavior of Python Image Library
         shape = tf.reshape(tf.stack([height, width, num_channels], axis=0),
                            shape=(3,), name='shape')
-        if img_dtype == SparkMode.RGB:
+        if img_dtype == 'uint8':
             image_uint8 = tf.decode_raw(image_buffer, tf.uint8, name="decode_raw")
             image_float = tf.to_float(image_uint8)
-        else:
-            assert img_dtype == SparkMode.RGB_FLOAT32, \
-                "Unsupported dtype for image: {}".format(img_dtype)
+        elif img_dtype == 'float32':
             image_float = tf.decode_raw(image_buffer, tf.float32, name="decode_raw")
-
+        else:
+            raise ValueError(
+                'unsupported image data type "%s", currently only know how to handle uint8 and float32' % img_dtype)
         image_reshaped = tf.reshape(image_float, shape, name="reshaped")
+        image_reshaped = imageIO.fixColorChannelOrdering(channelOrder, image_reshaped)
         image_input = tf.expand_dims(image_reshaped, 0, name="image_input")
         gfn = issn.asGraphFunction([height, width, image_buffer, num_channels], [image_input])
 
     return gfn
 
+
 def buildFlattener():
-    """ 
-    Build a flattening layer to remove the extra leading tensor dimension. 
+    """
+    Build a flattening layer to remove the extra leading tensor dimension.
     e.g. a tensor of shape [1, W, H, C] will have a shape [W, H, C] after applying this.
     """
     with IsolatedSession() as issn:

python/sparkdl/graph/tensorframes_udf.py

@@ -23,6 +23,7 @@
 
 logger = logging.getLogger('sparkdl')
 
+
 def makeGraphUDF(graph, udf_name, fetches, feeds_to_fields_map=None, blocked=False, register=True):
     """
     Create a Spark SQL UserDefinedFunction from a given TensorFlow Graph

python/sparkdl/graph/utils.py

@@ -31,6 +31,7 @@
 one of the four target variants.
 """
 
+
 def validated_graph(graph):
     """
     Check if the input is a valid :py:class:`tf.Graph` and return it.
@@ -41,6 +42,7 @@ def validated_graph(graph):
     assert isinstance(graph, tf.Graph), 'must provide tf.Graph, but get {}'.format(type(graph))
     return graph
 
+
 def get_shape(tfobj_or_name, graph):
     """
     Return the shape of the tensor as a list
@@ -52,6 +54,7 @@ def get_shape(tfobj_or_name, graph):
     _shape = get_tensor(tfobj_or_name, graph).get_shape().as_list()
     return [-1 if x is None else x for x in _shape]
 
+
 def get_op(tfobj_or_name, graph):
     """
     Get a :py:class:`tf.Operation` object.
@@ -76,6 +79,7 @@ def get_op(tfobj_or_name, graph):
     assert isinstance(op, tf.Operation), err_msg.format(_op_name, type(op), op)
     return op
 
+
 def get_tensor(tfobj_or_name, graph):
     """
     Get a :py:class:`tf.Tensor` object
@@ -100,6 +104,7 @@ def get_tensor(tfobj_or_name, graph):
     assert isinstance(tnsr, tf.Tensor), err_msg.format(_tensor_name, type(tnsr), tnsr)
     return tnsr
 
+
 def tensor_name(tfobj_or_name, graph=None):
     """
     Derive the :py:class:`tf.Tensor` name from a :py:class:`tf.Operation` or :py:class:`tf.Tensor`
@@ -130,6 +135,7 @@ def tensor_name(tfobj_or_name, graph=None):
     else:
         raise TypeError('invalid tf.Tensor name query type {}'.format(type(tfobj_or_name)))
 
+
 def op_name(tfobj_or_name, graph=None):
     """
     Derive the :py:class:`tf.Operation` name from a :py:class:`tf.Operation` or
@@ -158,9 +164,11 @@ def op_name(tfobj_or_name, graph=None):
     else:
         raise TypeError('invalid tf.Operation name query type {}'.format(type(tfobj_or_name)))
 
+
 def add_scope_to_name(scope, name):
     """ Prepends the provided scope to the passed-in op or tensor name. """
-    return "%s/%s"%(scope, name)
+    return "%s/%s" % (scope, name)
+
 
 def validated_output(tfobj_or_name, graph):
     """
@@ -172,6 +180,7 @@ def validated_output(tfobj_or_name, graph):
     graph = validated_graph(graph)
     return op_name(tfobj_or_name, graph)
 
+
 def validated_input(tfobj_or_name, graph):
     """
     Validate and return the input names useable GraphFunction
@@ -186,6 +195,7 @@ def validated_input(tfobj_or_name, graph):
         ('input must be Placeholder, but get', op.type)
     return name
 
+
 def strip_and_freeze_until(fetches, graph, sess=None, return_graph=False):
     """
     Create a static view of the graph by