broadcast only tunable parameters (#120)

* Broadcast only tunable parameters in KerasImageFileEstimator's fitMultiple implementation to mitigate issues where some untunable parameters (e.g. imageLoader) were unpicklable, causing the broadcast to fail.

Author: yogeshg

python/sparkdl/estimators/keras_image_file_estimator.py

@@ -148,6 +148,8 @@ def __init__(self, inputCol=None, outputCol=None, outputMode="vector", labelCol=
         super(KerasImageFileEstimator, self).__init__()
         kwargs = self._input_kwargs
         self.setParams(**kwargs)
+        self._tunable_params = [self.kerasOptimizer, self.kerasLoss, self.kerasFitParams,
+                                self.outputCol, self.outputMode]  # model params and output params
 
     @keyword_only
     def setParams(self, inputCol=None, outputCol=None, outputMode="vector", labelCol=None,
@@ -168,14 +170,11 @@ def _validateParams(self, paramMap):
         :return: True if parameters are valid
         """
 
-        tunable_params = [self.kerasOptimizer, self.kerasLoss, self.kerasFitParams,  # model params
-                          self.outputCol, self.outputMode]  # output params
-
         undefined = set([p for p in self.params if not self.isDefined(p)])
-        undefined_tunable = undefined.intersection(tunable_params)
+        undefined_tunable = undefined.intersection(self._tunable_params)
         failed_define = [p.name for p in undefined.difference(undefined_tunable)]
         failed_tune = [p.name for p in undefined_tunable if p not in paramMap]
-        untunable_overrides = [p.name for p in paramMap if p not in tunable_params]
+        untunable_overrides = [p.name for p in paramMap if p not in self._tunable_params]
 
         if failed_define or failed_tune or untunable_overrides:
             msg = "Following Params must be"
@@ -266,14 +265,16 @@ def fitMultiple(self, dataset, paramMaps):
         """
         [self._validateParams(pm) for pm in paramMaps]
 
-        def _name_value_map(paramMap):
-            """takes a dictionary {param -> value} and returns a map of {param.name -> value}"""
-            return {param.name: val for param, val in paramMap.items()}
+        def _get_tunable_name_value_map(param_map, tunable):
+            """takes a dictionary {`Param` -> value} and a list [`Param`], select keys that are
+            present in both and returns a map of {Param.name -> value}"""
+            return {param.name: val for param, val in param_map.items() if param in tunable}
 
         sc = JVMAPI._curr_sc()
-        paramNameMaps = list(enumerate(map(_name_value_map, paramMaps)))
-        num_models = len(paramNameMaps)
-        paramNameMapsRDD = sc.parallelize(paramNameMaps, numSlices=num_models)
+        param_name_maps = [(i, _get_tunable_name_value_map(pm, self._tunable_params))
+                           for (i, pm) in enumerate(paramMaps)]
+        num_models = len(param_name_maps)
+        paramNameMapsRDD = sc.parallelize(param_name_maps, numSlices=num_models)
 
         # Extract image URI from provided dataset and create features as numpy arrays
         localFeatures, localLabels = self._getNumpyFeaturesAndLabels(dataset)
@@ -285,8 +286,8 @@ def _name_value_map(paramMap):
         modelBytesBc = sc.broadcast(modelBytes)
 
         # Obtain params for this estimator instance
-        baseParams = _name_value_map(self.extractParamMap())
-        baseParamsBc = sc.broadcast(baseParams)
+        base_params = _get_tunable_name_value_map(self.extractParamMap(), self._tunable_params)
+        baseParamsBc = sc.broadcast(base_params)
 
         def _local_fit(row):
             """