Repository: mlflow/mlflow
Pull request: Fix Spark bug loading Spark UDFs on Databricks
## What changes are proposed in this pull request?
 
Fixes #1657 
 
## How is this patch tested?
 
Existing unit tests & manual testing.
 
## Release Notes
 
### Is this a user-facing change? 

- [ ] No. You can skip the rest of this section.
- [x] Yes. Give a description of this change to be included in the release notes for MLflow users.
 
Fixes a bug in MLFlow 1.1.0 where loading MLflow models as Spark UDFs on Databricks would fail with an obscure `ValueError: ZIP does not support timestamps before 1980` error message
 
### What component(s) does this PR affect?
 
- [ ] UI
- [ ] CLI 
- [ ] API 
- [ ] REST-API 
- [ ] Examples 
- [ ] Docs
- [ ] Tracking
- [ ] Projects 
- [ ] Artifacts 
- [x] Models 
- [ ] Scoring 
- [ ] Serving
- [ ] R
- [ ] Java
- [ ] Python

### How should the PR be classified in the release notes? Choose one:
 
- [ ] `rn/breaking-change` - The PR will be mentioned in the "Breaking Changes" section
- [ ] `rn/none` - No description will be included. The PR will be mentioned only by the PR number in the "Small Bugfixes and Documentation Updates" section
- [ ] `rn/feature` - A new user-facing feature worth mentioning in the release notes
- [x] `rn/bug-fix` - A user-facing bug fix worth mentioning in the release notes
- [ ] `rn/documentation` - A user-facing documentation change worth mentioning in the release notes


Problem statement:
[MLflow 1.1] Loading models with mlflow.pyfunc.spark_udf broken on Databricks clusters
### System information
- **Have I written custom code (as opposed to using a stock example script provided in MLflow)**: No
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Ubuntu 16.04
- **MLflow installed from (source or binary)**:  Binary
- **MLflow version (run ``mlflow --version``)**: 1.1.0
- **Python version**: 3.6
- **npm version (if running the dev UI): N/A
- **Exact command to reproduce**: Try to load a model as a Spark UDF via `mlflow.pyfunc.spark_udf`

### Describe the problem
Loading models using a Spark UDF is broken in MLflow 1.1 due to a bug in zipping files from DBFS FUSE (the timestamps on the files are misconfigured or read incorrectly).

As a workaround, you can disable the DBFS FUSE logic before loading your model via:
```
import os
os.environ["MLFLOW_ENABLE_DBFS_FUSE_ARTIFACT_REPO"] = 'false'
```


### Source code / logs
```
---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
<command-3584523> in <module>()
     1 import mlflow.pyfunc
     2
----> 3 trained_model = mlflow.pyfunc.spark_udf(spark, model_uri='runs:/c4569266b782467cacd91c3d0ef8f9fc/model')

/databricks/python/lib/python3.6/site-packages/mlflow/pyfunc/__init__.py in spark_udf(spark, model_uri, result_type)
   416
   417     local_model_path = _download_artifact_from_uri(artifact_uri=model_uri)
--> 418     archive_path = SparkModelCache.add_local_model(spark, local_model_path)
   419
   420     def predict(*args):

/databricks/python/lib/python3.6/site-packages/mlflow/pyfunc/spark_model_cache.py in add_local_model(spark, model_path)
    35         # NB: We must archive the directory as Spark.addFile does not support non-DFS
    36         # directories when recursive=True.
---> 37         archive_path = shutil.make_archive(archive_basepath, 'zip', model_path)
    38         spark.sparkContext.addFile(archive_path)
    39         return archive_path

/databricks/python/lib/python3.6/shutil.py in make_archive(base_name, format, root_dir, base_dir, verbose, dry_run, owner, group, logger)
   798
   799     try:
--> 800         filename = func(base_name, base_dir, **kwargs)
   801     finally:
   802         if root_dir is not None:

/databricks/python/lib/python3.6/shutil.py in _make_zipfile(base_name, base_dir, verbose, dry_run, logger)
   690                 for name in sorted(dirnames):
   691                     path = os.path.normpath(os.path.join(dirpath, name))
--> 692                     zf.write(path, path)
   693                     if logger is not None:
   694                         logger.info("adding '%s'", path)

/databricks/python/lib/python3.6/zipfile.py in write(self, filename, arcname, compress_type)
  1592             )
  1593
-> 1594         zinfo = ZipInfo.from_file(filename, arcname)
  1595
  1596         if zinfo.is_dir():

/databricks/python/lib/python3.6/zipfile.py in from_file(cls, filename, arcname)
   494         if isdir:
   495             arcname += '/'
--> 496         zinfo = cls(arcname, date_time)
   497         zinfo.external_attr = (st.st_mode & 0xFFFF) << 16  # Unix attributes
   498         if isdir:

/databricks/python/lib/python3.6/zipfile.py in __init__(self, filename, date_time)
   336
   337         if date_time[0] < 1980:
--> 338             raise ValueError('ZIP does not support timestamps before 1980')
   339
   340         # Standard values:

ValueError: ZIP does not support timestamps before 1980
```

Hints from issue discussion:
I noticed this bug with Keras/TensorFlow on DBR ML 5.4. However  mlflow.pyfunc.spark_udf() with sklearn works fine.

Commit message: WIP, fix test

Patch to review:
diff --git a/mlflow/pyfunc/__init__.py b/mlflow/pyfunc/__init__.py
index 58ba8d8bf9c88..52e4aeb124fab 100644
--- a/mlflow/pyfunc/__init__.py
+++ b/mlflow/pyfunc/__init__.py
@@ -414,8 +414,10 @@ def spark_udf(spark, model_uri, result_type="double"):
                     "of the following types types: {}".format(str(elem_type), str(supported_types)),
             error_code=INVALID_PARAMETER_VALUE)
 
-    local_model_path = _download_artifact_from_uri(artifact_uri=model_uri)
-    archive_path = SparkModelCache.add_local_model(spark, local_model_path)
+    with TempDir() as local_tmpdir:
+        local_model_path = _download_artifact_from_uri(
+            artifact_uri=model_uri, output_path=local_tmpdir.path())
+        archive_path = SparkModelCache.add_local_model(spark, local_model_path)
 
     def predict(*args):
         model = SparkModelCache.get_or_load(archive_path)
diff --git a/mlflow/store/local_artifact_repo.py b/mlflow/store/local_artifact_repo.py
index 69d836c488376..410d218e2d812 100644
--- a/mlflow/store/local_artifact_repo.py
+++ b/mlflow/store/local_artifact_repo.py
@@ -72,7 +72,7 @@ def list_artifacts(self, path=None):
             path = os.path.normpath(path)
         list_dir = os.path.join(self.artifact_dir, path) if path else self.artifact_dir
         if os.path.isdir(list_dir):
-            artifact_files = list_all(list_dir, full_path=True)
+            artifact_files = list_all(list_dir, full_path=True) + [list_dir]
             infos = [get_file_info(f,
                                    relative_path_to_artifact_path(
                                        os.path.relpath(f, self.artifact_dir)))


Code review comments:
diff_hunk: @@ -72,7 +72,7 @@ def list_artifacts(self, path=None):
             path = os.path.normpath(path)
         list_dir = os.path.join(self.artifact_dir, path) if path else self.artifact_dir
         if os.path.isdir(list_dir):
-            artifact_files = list_all(list_dir, full_path=True)
+            artifact_files = list_all(list_dir, full_path=True) + [list_dir]
path: mlflow/store/local_artifact_repo.py
text: @author:
The spark UDF test (`pytest tests/pyfunc/test_spark.py::test_spark_udf --large`) was failing due to leaked mlruns directory from another test run (we should fix that separately), see [this travis log](https://travis-ci.org/mlflow/mlflow/jobs/562924587). The failure revealed that we currently have a bug/edge case in how we handle downloading artifacts from empty directories (in ArtifactRepository.download_artifacts, we assume that an empty result from list_artifacts implies that an artifact path is a file, but it could actually be an empty directory, e.g. the `mlruns` directory in the above failure). I changed this code to get the test to pass on the PR, but this is probably not the right solution (will think about this more tomorrow)

@user1:
I don't understand why do you want to add the list_dir to the result. Can you elaborate on what caused the error and how does this fixes it? From looking at it briefly I don't think the list_dir should be returned. If anything, I don't understand why list artifacts pointing to a file does not return the file itself. 

@author:
Yeah agreed, adding `list_dir` is just a hack to get the tests to pass against the file-based artifact repo, because it allows us to distinguish between listing an empty directory & a file.

Agree that it'd make sense for list_artifacts pointing a file to return the file itself (similar to how `ls` behaves), but that's a broader change that requires modifications across the different artifact repository implementations, so I just worked around this for now by removing the problematic empty `mlruns` directory in the `projects` tests that create it.