We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] HdfsArtifactRepository list_artifacts recursively lists all items under directory
Thank you for submitting an issue. Please refer to our [issue policy](https://www.github.com/mlflow/mlflow/blob/master/ISSUE_POLICY.md)
for information on what types of issues we address.

For help with debugging your code, please refer to [Stack Overflow](https://stackoverflow.com/questions/tagged/mlflow).

  
Please do not delete this template unless you are sure your issue is outside its scope.

### 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)**: Centos 7
- **MLflow installed from (source or binary)**: binary
- **MLflow version (run ``mlflow --version``)**: 1.1.0
- **Python version**: 3.6.8
- **npm version, if running the dev UI**: N/A
- **Exact command to reproduce**: mlflow artifacts list -r <run id for artifacts stored on hdfs>

### Describe the problem
list_artifacts of an artifact repository is expected to only list the files directly under the provided path (see https://github.com/mlflow/mlflow/blob/4b1868719837d1844f19b6242643222549ee2794/mlflow/store/cli.py#L74 ). HdfsArtifactRepository walks all files under the given path and returns them (see https://github.com/mlflow/mlflow/blob/4b1868719837d1844f19b6242643222549ee2794/mlflow/store/hdfs_artifact_repo.py#L89 ).

This behavior breaks the mflow server as it expects the behavior specified in the cli file.

### Code to reproduce issue
Provide a reproducible test case that is the bare minimum necessary to generate the problem.

### Other info / logs
Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks,
please include the full traceback. Large logs and files should be attached.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mlflow/store/hdfs_artifact_repo.py`
Content:
```
1 import os
2 import posixpath
3 import tempfile
4 from contextlib import contextmanager
5 
6 from six.moves import urllib
7 
8 from mlflow.entities import FileInfo
9 from mlflow.exceptions import MlflowException
10 from mlflow.store.artifact_repo import ArtifactRepository
11 from mlflow.utils.file_utils import mkdir, relative_path_to_artifact_path
12 
13 
14 class HdfsArtifactRepository(ArtifactRepository):
15     """
16     Stores artifacts on HDFS.
17 
18     This repository is used with URIs of the form ``hdfs:/<path>``. The repository can only be used
19     together with the RestStore.
20     """
21 
22     def __init__(self, artifact_uri):
23         self.host, self.port, self.path = _resolve_connection_params(artifact_uri)
24         super(HdfsArtifactRepository, self).__init__(artifact_uri)
25 
26     def log_artifact(self, local_file, artifact_path=None):
27         """
28             Log artifact in hdfs.
29         :param local_file: source file path
30         :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path
31         """
32         hdfs_base_path = _resolve_base_path(self.path, artifact_path)
33 
34         with hdfs_system(host=self.host, port=self.port) as hdfs:
35             _, file_name = os.path.split(local_file)
36             destination = posixpath.join(hdfs_base_path, file_name)
37             with hdfs.open(destination, 'wb') as output:
38                 output.write(open(local_file, "rb").read())
39 
40     def log_artifacts(self, local_dir, artifact_path=None):
41         """
42             Log artifacts in hdfs.
43             Missing remote sub-directories will be created if needed.
44         :param local_dir: source dir path
45         :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path
46         """
47         hdfs_base_path = _resolve_base_path(self.path, artifact_path)
48 
49         with hdfs_system(host=self.host, port=self.port) as hdfs:
50 
51             if not hdfs.exists(hdfs_base_path):
52                 hdfs.mkdir(hdfs_base_path)
53 
54             for subdir_path, _, files in os.walk(local_dir):
55 
56                 relative_path = _relative_path_local(local_dir, subdir_path)
57 
58                 hdfs_subdir_path = posixpath.join(hdfs_base_path, relative_path) \
59                     if relative_path else hdfs_base_path
60 
61                 if not hdfs.exists(hdfs_subdir_path):
62                     hdfs.mkdir(hdfs_subdir_path)
63 
64                 for each_file in files:
65                     source = os.path.join(subdir_path, each_file)
66                     destination = posixpath.join(hdfs_subdir_path, each_file)
67                     with hdfs.open(destination, 'wb') as output_stream:
68                         output_stream.write(open(source, "rb").read())
69 
70     def list_artifacts(self, path=None):
71         """
72             Lists files and directories under artifacts directory for the current run_id.
73             (self.path contains the base path - hdfs:/some/path/run_id/artifacts)
74 
75             :param path: Relative source path. Possible subdirectory existing under
76                          hdfs:/some/path/run_id/artifacts
77             :return: List of files and directories under given path -
78                 example:
79                     ['conda.yaml', 'MLmodel', 'model.pkl']
80         """
81         hdfs_base_path = _resolve_base_path(self.path, path)
82         base_path_len = len(hdfs_base_path) + 1
83 
84         with hdfs_system(host=self.host, port=self.port) as hdfs:
85             paths = []
86             for path, is_dir, size in self._walk_path(hdfs, hdfs_base_path):
87                 paths.append(FileInfo(path[base_path_len:], is_dir, size))
88             return sorted(paths, key=lambda f: paths)
89 
90     def _walk_path(self, hdfs, hdfs_path):
91         if hdfs.exists(hdfs_path):
92             if hdfs.isdir(hdfs_path):
93                 for subdir, _, files in hdfs.walk(hdfs_path):
94                     if subdir != hdfs_path:
95                         yield subdir, hdfs.isdir(subdir), hdfs.info(subdir).get("size")
96                     for f in files:
97                         file_path = posixpath.join(subdir, f)
98                         yield file_path, hdfs.isdir(file_path), hdfs.info(file_path).get("size")
99             else:
100                 yield hdfs_path, False, hdfs.info(hdfs_path).get("size")
101 
102     def download_artifacts(self, artifact_path, dst_path=None):
103         """
104             Download an artifact file or directory to a local directory/file if applicable, and
105             return a local path for it.
106             The caller is responsible for managing the lifecycle of the downloaded artifacts.
107 
108             (self.path contains the base path - hdfs:/some/path/run_id/artifacts)
109 
110             :param artifact_path: Relative source path to the desired artifacts file or directory.
111             :param dst_path: Absolute path of the local filesystem destination directory to which
112                              to download the specified artifacts. This directory must already
113                              exist. If unspecified, the artifacts will be downloaded to a new,
114                              uniquely-named
115                              directory on the local filesystem.
116 
117             :return: Absolute path of the local filesystem location containing the downloaded
118             artifacts - file/directory.
119         """
120 
121         hdfs_base_path = _resolve_base_path(self.path, artifact_path)
122         local_dir = _tmp_dir(dst_path)
123 
124         with hdfs_system(host=self.host, port=self.port) as hdfs:
125 
126             if not hdfs.isdir(hdfs_base_path):
127                 local_path = os.path.join(local_dir, os.path.normpath(artifact_path))
128                 _download_hdfs_file(hdfs, hdfs_base_path, local_path)
129                 return local_path
130 
131             for path, is_dir, _ in self._walk_path(hdfs, hdfs_base_path):
132 
133                 relative_path = _relative_path_remote(hdfs_base_path, path)
134                 local_path = os.path.join(local_dir, relative_path) \
135                     if relative_path else local_dir
136 
137                 if is_dir:
138                     mkdir(local_path)
139                 else:
140                     _download_hdfs_file(hdfs, path, local_path)
141             return local_dir
142 
143     def _download_file(self, remote_file_path, local_path):
144         raise MlflowException('This is not implemented. Should never be called.')
145 
146 
147 @contextmanager
148 def hdfs_system(host, port):
149     """
150         hdfs system context - Attempt to establish the connection to hdfs
151         and yields HadoopFileSystem
152 
153     :param host: hostname or when relaying on the core-site.xml config use 'default'
154     :param port: port or when relaying on the core-site.xml config use 0
155     """
156     import pyarrow as pa
157 
158     driver = os.getenv('MLFLOW_HDFS_DRIVER') or 'libhdfs'
159     kerb_ticket = os.getenv('MLFLOW_KERBEROS_TICKET_CACHE')
160     kerberos_user = os.getenv('MLFLOW_KERBEROS_USER')
161     extra_conf = _parse_extra_conf(os.getenv('MLFLOW_PYARROW_EXTRA_CONF'))
162 
163     connected = pa.hdfs.connect(host=host or 'default',
164                                 port=port or 0,
165                                 user=kerberos_user,
166                                 driver=driver,
167                                 kerb_ticket=kerb_ticket,
168                                 extra_conf=extra_conf)
169     yield connected
170     connected.close()
171 
172 
173 def _resolve_connection_params(artifact_uri):
174     parsed = urllib.parse.urlparse(artifact_uri)
175     return parsed.hostname, parsed.port, parsed.path
176 
177 
178 def _resolve_base_path(path, artifact_path):
179     if path == artifact_path:
180         return path
181     if artifact_path:
182         return posixpath.join(path, artifact_path)
183     return path
184 
185 
186 def _relative_path(base_dir, subdir_path, path_module):
187     relative_path = path_module.relpath(subdir_path, base_dir)
188     return relative_path if relative_path is not '.' else None
189 
190 
191 def _relative_path_local(base_dir, subdir_path):
192     rel_path = _relative_path(base_dir, subdir_path, os.path)
193     return relative_path_to_artifact_path(rel_path) if rel_path is not None else None
194 
195 
196 def _relative_path_remote(base_dir, subdir_path):
197     return _relative_path(base_dir, subdir_path, posixpath)
198 
199 
200 def _tmp_dir(local_path):
201     return os.path.abspath(tempfile.mkdtemp(dir=local_path))
202 
203 
204 def _download_hdfs_file(hdfs, remote_file_path, local_file_path):
205     with open(local_file_path, 'wb') as f:
206         f.write(hdfs.open(remote_file_path, 'rb').read())
207 
208 
209 def _parse_extra_conf(extra_conf):
210     if extra_conf:
211         def as_pair(config):
212             key, val = config.split('=')
213             return key, val
214 
215         list_of_key_val = [as_pair(conf) for conf in extra_conf.split(',')]
216         return dict(list_of_key_val)
217     return None
218
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/mlflow/store/hdfs_artifact_repo.py b/mlflow/store/hdfs_artifact_repo.py
--- a/mlflow/store/hdfs_artifact_repo.py
+++ b/mlflow/store/hdfs_artifact_repo.py
@@ -74,17 +74,21 @@
 
             :param path: Relative source path. Possible subdirectory existing under
                          hdfs:/some/path/run_id/artifacts
-            :return: List of files and directories under given path -
-                example:
-                    ['conda.yaml', 'MLmodel', 'model.pkl']
+            :return: List of FileInfos under given path
         """
         hdfs_base_path = _resolve_base_path(self.path, path)
-        base_path_len = len(hdfs_base_path) + 1
 
         with hdfs_system(host=self.host, port=self.port) as hdfs:
             paths = []
-            for path, is_dir, size in self._walk_path(hdfs, hdfs_base_path):
-                paths.append(FileInfo(path[base_path_len:], is_dir, size))
+            if hdfs.exists(hdfs_base_path):
+                for file_detail in hdfs.ls(hdfs_base_path, detail=True):
+                    file_name = file_detail.get("name")
+                    # Strip off anything that comes before the artifact root e.g. hdfs://name
+                    offset = file_name.index(self.path)
+                    rel_path = _relative_path_remote(self.path, file_name[offset:])
+                    is_dir = file_detail.get("kind") == "directory"
+                    size = file_detail.get("size")
+                    paths.append(FileInfo(rel_path, is_dir, size))
             return sorted(paths, key=lambda f: paths)
 
     def _walk_path(self, hdfs, hdfs_path):
@@ -202,6 +206,9 @@
 
 
 def _download_hdfs_file(hdfs, remote_file_path, local_file_path):
+    # Ensure all required directories exist. Without doing this nested files can't be downloaded.
+    dirs = os.path.dirname(local_file_path)
+    os.makedirs(dirs)
     with open(local_file_path, 'wb') as f:
         f.write(hdfs.open(remote_file_path, 'rb').read())

{"golden_diff": "diff --git a/mlflow/store/hdfs_artifact_repo.py b/mlflow/store/hdfs_artifact_repo.py\n--- a/mlflow/store/hdfs_artifact_repo.py\n+++ b/mlflow/store/hdfs_artifact_repo.py\n@@ -74,17 +74,21 @@\n \n             :param path: Relative source path. Possible subdirectory existing under\n                          hdfs:/some/path/run_id/artifacts\n-            :return: List of files and directories under given path -\n-                example:\n-                    ['conda.yaml', 'MLmodel', 'model.pkl']\n+            :return: List of FileInfos under given path\n         \"\"\"\n         hdfs_base_path = _resolve_base_path(self.path, path)\n-        base_path_len = len(hdfs_base_path) + 1\n \n         with hdfs_system(host=self.host, port=self.port) as hdfs:\n             paths = []\n-            for path, is_dir, size in self._walk_path(hdfs, hdfs_base_path):\n-                paths.append(FileInfo(path[base_path_len:], is_dir, size))\n+            if hdfs.exists(hdfs_base_path):\n+                for file_detail in hdfs.ls(hdfs_base_path, detail=True):\n+                    file_name = file_detail.get(\"name\")\n+                    # Strip off anything that comes before the artifact root e.g. hdfs://name\n+                    offset = file_name.index(self.path)\n+                    rel_path = _relative_path_remote(self.path, file_name[offset:])\n+                    is_dir = file_detail.get(\"kind\") == \"directory\"\n+                    size = file_detail.get(\"size\")\n+                    paths.append(FileInfo(rel_path, is_dir, size))\n             return sorted(paths, key=lambda f: paths)\n \n     def _walk_path(self, hdfs, hdfs_path):\n@@ -202,6 +206,9 @@\n \n \n def _download_hdfs_file(hdfs, remote_file_path, local_file_path):\n+    # Ensure all required directories exist. Without doing this nested files can't be downloaded.\n+    dirs = os.path.dirname(local_file_path)\n+    os.makedirs(dirs)\n     with open(local_file_path, 'wb') as f:\n         f.write(hdfs.open(remote_file_path, 'rb').read())\n", "issue": "[BUG] HdfsArtifactRepository list_artifacts recursively lists all items under directory\nThank you for submitting an issue. Please refer to our [issue policy](https://www.github.com/mlflow/mlflow/blob/master/ISSUE_POLICY.md)\r\nfor information on what types of issues we address.\r\n\r\nFor help with debugging your code, please refer to [Stack Overflow](https://stackoverflow.com/questions/tagged/mlflow).\r\n\r\n  \r\nPlease do not delete this template unless you are sure your issue is outside its scope.\r\n\r\n### System information\r\n- **Have I written custom code (as opposed to using a stock example script provided in MLflow)**: No\r\n- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Centos 7\r\n- **MLflow installed from (source or binary)**: binary\r\n- **MLflow version (run ``mlflow --version``)**: 1.1.0\r\n- **Python version**: 3.6.8\r\n- **npm version, if running the dev UI**: N/A\r\n- **Exact command to reproduce**: mlflow artifacts list -r <run id for artifacts stored on hdfs>\r\n\r\n### Describe the problem\r\nlist_artifacts of an artifact repository is expected to only list the files directly under the provided path (see https://github.com/mlflow/mlflow/blob/4b1868719837d1844f19b6242643222549ee2794/mlflow/store/cli.py#L74 ). HdfsArtifactRepository walks all files under the given path and returns them (see https://github.com/mlflow/mlflow/blob/4b1868719837d1844f19b6242643222549ee2794/mlflow/store/hdfs_artifact_repo.py#L89 ).\r\n\r\nThis behavior breaks the mflow server as it expects the behavior specified in the cli file.\r\n\r\n### Code to reproduce issue\r\nProvide a reproducible test case that is the bare minimum necessary to generate the problem.\r\n\r\n### Other info / logs\r\nInclude any logs or source code that would be helpful to diagnose the problem. If including tracebacks,\r\nplease include the full traceback. Large logs and files should be attached.\r\n\n", "before_files": [{"content": "import os\nimport posixpath\nimport tempfile\nfrom contextlib import contextmanager\n\nfrom six.moves import urllib\n\nfrom mlflow.entities import FileInfo\nfrom mlflow.exceptions import MlflowException\nfrom mlflow.store.artifact_repo import ArtifactRepository\nfrom mlflow.utils.file_utils import mkdir, relative_path_to_artifact_path\n\n\nclass HdfsArtifactRepository(ArtifactRepository):\n    \"\"\"\n    Stores artifacts on HDFS.\n\n    This repository is used with URIs of the form ``hdfs:/<path>``. The repository can only be used\n    together with the RestStore.\n    \"\"\"\n\n    def __init__(self, artifact_uri):\n        self.host, self.port, self.path = _resolve_connection_params(artifact_uri)\n        super(HdfsArtifactRepository, self).__init__(artifact_uri)\n\n    def log_artifact(self, local_file, artifact_path=None):\n        \"\"\"\n            Log artifact in hdfs.\n        :param local_file: source file path\n        :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n\n        with hdfs_system(host=self.host, port=self.port) as hdfs:\n            _, file_name = os.path.split(local_file)\n            destination = posixpath.join(hdfs_base_path, file_name)\n            with hdfs.open(destination, 'wb') as output:\n                output.write(open(local_file, \"rb\").read())\n\n    def log_artifacts(self, local_dir, artifact_path=None):\n        \"\"\"\n            Log artifacts in hdfs.\n            Missing remote sub-directories will be created if needed.\n        :param local_dir: source dir path\n        :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n\n        with hdfs_system(host=self.host, port=self.port) as hdfs:\n\n            if not hdfs.exists(hdfs_base_path):\n                hdfs.mkdir(hdfs_base_path)\n\n            for subdir_path, _, files in os.walk(local_dir):\n\n                relative_path = _relative_path_local(local_dir, subdir_path)\n\n                hdfs_subdir_path = posixpath.join(hdfs_base_path, relative_path) \\\n                    if relative_path else hdfs_base_path\n\n                if not hdfs.exists(hdfs_subdir_path):\n                    hdfs.mkdir(hdfs_subdir_path)\n\n                for each_file in files:\n                    source = os.path.join(subdir_path, each_file)\n                    destination = posixpath.join(hdfs_subdir_path, each_file)\n                    with hdfs.open(destination, 'wb') as output_stream:\n                        output_stream.write(open(source, \"rb\").read())\n\n    def list_artifacts(self, path=None):\n        \"\"\"\n            Lists files and directories under artifacts directory for the current run_id.\n            (self.path contains the base path - hdfs:/some/path/run_id/artifacts)\n\n            :param path: Relative source path. Possible subdirectory existing under\n                         hdfs:/some/path/run_id/artifacts\n            :return: List of files and directories under given path -\n                example:\n                    ['conda.yaml', 'MLmodel', 'model.pkl']\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, path)\n        base_path_len = len(hdfs_base_path) + 1\n\n        with hdfs_system(host=self.host, port=self.port) as hdfs:\n            paths = []\n            for path, is_dir, size in self._walk_path(hdfs, hdfs_base_path):\n                paths.append(FileInfo(path[base_path_len:], is_dir, size))\n            return sorted(paths, key=lambda f: paths)\n\n    def _walk_path(self, hdfs, hdfs_path):\n        if hdfs.exists(hdfs_path):\n            if hdfs.isdir(hdfs_path):\n                for subdir, _, files in hdfs.walk(hdfs_path):\n                    if subdir != hdfs_path:\n                        yield subdir, hdfs.isdir(subdir), hdfs.info(subdir).get(\"size\")\n                    for f in files:\n                        file_path = posixpath.join(subdir, f)\n                        yield file_path, hdfs.isdir(file_path), hdfs.info(file_path).get(\"size\")\n            else:\n                yield hdfs_path, False, hdfs.info(hdfs_path).get(\"size\")\n\n    def download_artifacts(self, artifact_path, dst_path=None):\n        \"\"\"\n            Download an artifact file or directory to a local directory/file if applicable, and\n            return a local path for it.\n            The caller is responsible for managing the lifecycle of the downloaded artifacts.\n\n            (self.path contains the base path - hdfs:/some/path/run_id/artifacts)\n\n            :param artifact_path: Relative source path to the desired artifacts file or directory.\n            :param dst_path: Absolute path of the local filesystem destination directory to which\n                             to download the specified artifacts. This directory must already\n                             exist. If unspecified, the artifacts will be downloaded to a new,\n                             uniquely-named\n                             directory on the local filesystem.\n\n            :return: Absolute path of the local filesystem location containing the downloaded\n            artifacts - file/directory.\n        \"\"\"\n\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n        local_dir = _tmp_dir(dst_path)\n\n        with hdfs_system(host=self.host, port=self.port) as hdfs:\n\n            if not hdfs.isdir(hdfs_base_path):\n                local_path = os.path.join(local_dir, os.path.normpath(artifact_path))\n                _download_hdfs_file(hdfs, hdfs_base_path, local_path)\n                return local_path\n\n            for path, is_dir, _ in self._walk_path(hdfs, hdfs_base_path):\n\n                relative_path = _relative_path_remote(hdfs_base_path, path)\n                local_path = os.path.join(local_dir, relative_path) \\\n                    if relative_path else local_dir\n\n                if is_dir:\n                    mkdir(local_path)\n                else:\n                    _download_hdfs_file(hdfs, path, local_path)\n            return local_dir\n\n    def _download_file(self, remote_file_path, local_path):\n        raise MlflowException('This is not implemented. Should never be called.')\n\n\n@contextmanager\ndef hdfs_system(host, port):\n    \"\"\"\n        hdfs system context - Attempt to establish the connection to hdfs\n        and yields HadoopFileSystem\n\n    :param host: hostname or when relaying on the core-site.xml config use 'default'\n    :param port: port or when relaying on the core-site.xml config use 0\n    \"\"\"\n    import pyarrow as pa\n\n    driver = os.getenv('MLFLOW_HDFS_DRIVER') or 'libhdfs'\n    kerb_ticket = os.getenv('MLFLOW_KERBEROS_TICKET_CACHE')\n    kerberos_user = os.getenv('MLFLOW_KERBEROS_USER')\n    extra_conf = _parse_extra_conf(os.getenv('MLFLOW_PYARROW_EXTRA_CONF'))\n\n    connected = pa.hdfs.connect(host=host or 'default',\n                                port=port or 0,\n                                user=kerberos_user,\n                                driver=driver,\n                                kerb_ticket=kerb_ticket,\n                                extra_conf=extra_conf)\n    yield connected\n    connected.close()\n\n\ndef _resolve_connection_params(artifact_uri):\n    parsed = urllib.parse.urlparse(artifact_uri)\n    return parsed.hostname, parsed.port, parsed.path\n\n\ndef _resolve_base_path(path, artifact_path):\n    if path == artifact_path:\n        return path\n    if artifact_path:\n        return posixpath.join(path, artifact_path)\n    return path\n\n\ndef _relative_path(base_dir, subdir_path, path_module):\n    relative_path = path_module.relpath(subdir_path, base_dir)\n    return relative_path if relative_path is not '.' else None\n\n\ndef _relative_path_local(base_dir, subdir_path):\n    rel_path = _relative_path(base_dir, subdir_path, os.path)\n    return relative_path_to_artifact_path(rel_path) if rel_path is not None else None\n\n\ndef _relative_path_remote(base_dir, subdir_path):\n    return _relative_path(base_dir, subdir_path, posixpath)\n\n\ndef _tmp_dir(local_path):\n    return os.path.abspath(tempfile.mkdtemp(dir=local_path))\n\n\ndef _download_hdfs_file(hdfs, remote_file_path, local_file_path):\n    with open(local_file_path, 'wb') as f:\n        f.write(hdfs.open(remote_file_path, 'rb').read())\n\n\ndef _parse_extra_conf(extra_conf):\n    if extra_conf:\n        def as_pair(config):\n            key, val = config.split('=')\n            return key, val\n\n        list_of_key_val = [as_pair(conf) for conf in extra_conf.split(',')]\n        return dict(list_of_key_val)\n    return None\n", "path": "mlflow/store/hdfs_artifact_repo.py"}], "after_files": [{"content": "import os\nimport posixpath\nimport tempfile\nfrom contextlib import contextmanager\n\nfrom six.moves import urllib\n\nfrom mlflow.entities import FileInfo\nfrom mlflow.exceptions import MlflowException\nfrom mlflow.store.artifact_repo import ArtifactRepository\nfrom mlflow.utils.file_utils import mkdir, relative_path_to_artifact_path\n\n\nclass HdfsArtifactRepository(ArtifactRepository):\n    \"\"\"\n    Stores artifacts on HDFS.\n\n    This repository is used with URIs of the form ``hdfs:/<path>``. The repository can only be used\n    together with the RestStore.\n    \"\"\"\n\n    def __init__(self, artifact_uri):\n        self.host, self.port, self.path = _resolve_connection_params(artifact_uri)\n        super(HdfsArtifactRepository, self).__init__(artifact_uri)\n\n    def log_artifact(self, local_file, artifact_path=None):\n        \"\"\"\n            Log artifact in hdfs.\n        :param local_file: source file path\n        :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n\n        with hdfs_system(host=self.host, port=self.port) as hdfs:\n            _, file_name = os.path.split(local_file)\n            destination = posixpath.join(hdfs_base_path, file_name)\n            with hdfs.open(destination, 'wb') as output:\n                output.write(open(local_file, \"rb\").read())\n\n    def log_artifacts(self, local_dir, artifact_path=None):\n        \"\"\"\n            Log artifacts in hdfs.\n            Missing remote sub-directories will be created if needed.\n        :param local_dir: source dir path\n        :param artifact_path: when specified will attempt to write under artifact_uri/artifact_path\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n\n        with hdfs_system(host=self.host, port=self.port) as hdfs:\n\n            if not hdfs.exists(hdfs_base_path):\n                hdfs.mkdir(hdfs_base_path)\n\n            for subdir_path, _, files in os.walk(local_dir):\n\n                relative_path = _relative_path_local(local_dir, subdir_path)\n\n                hdfs_subdir_path = posixpath.join(hdfs_base_path, relative_path) \\\n                    if relative_path else hdfs_base_path\n\n                if not hdfs.exists(hdfs_subdir_path):\n                    hdfs.mkdir(hdfs_subdir_path)\n\n                for each_file in files:\n                    source = os.path.join(subdir_path, each_file)\n                    destination = posixpath.join(hdfs_subdir_path, each_file)\n                    with hdfs.open(destination, 'wb') as output_stream:\n                        output_stream.write(open(source, \"rb\").read())\n\n    def list_artifacts(self, path=None):\n        \"\"\"\n            Lists files and directories under artifacts directory for the current run_id.\n            (self.path contains the base path - hdfs:/some/path/run_id/artifacts)\n\n            :param path: Relative source path. Possible subdirectory existing under\n                         hdfs:/some/path/run_id/artifacts\n            :return: List of FileInfos under given path\n        \"\"\"\n        hdfs_base_path = _resolve_base_path(self.path, path)\n\n        with hdfs_system(host=self.host, port=self.port) as hdfs:\n            paths = []\n            if hdfs.exists(hdfs_base_path):\n                for file_detail in hdfs.ls(hdfs_base_path, detail=True):\n                    file_name = file_detail.get(\"name\")\n                    # Strip off anything that comes before the artifact root e.g. hdfs://name\n                    offset = file_name.index(self.path)\n                    rel_path = _relative_path_remote(self.path, file_name[offset:])\n                    is_dir = file_detail.get(\"kind\") == \"directory\"\n                    size = file_detail.get(\"size\")\n                    paths.append(FileInfo(rel_path, is_dir, size))\n            return sorted(paths, key=lambda f: paths)\n\n    def _walk_path(self, hdfs, hdfs_path):\n        if hdfs.exists(hdfs_path):\n            if hdfs.isdir(hdfs_path):\n                for subdir, _, files in hdfs.walk(hdfs_path):\n                    if subdir != hdfs_path:\n                        yield subdir, hdfs.isdir(subdir), hdfs.info(subdir).get(\"size\")\n                    for f in files:\n                        file_path = posixpath.join(subdir, f)\n                        yield file_path, hdfs.isdir(file_path), hdfs.info(file_path).get(\"size\")\n            else:\n                yield hdfs_path, False, hdfs.info(hdfs_path).get(\"size\")\n\n    def download_artifacts(self, artifact_path, dst_path=None):\n        \"\"\"\n            Download an artifact file or directory to a local directory/file if applicable, and\n            return a local path for it.\n            The caller is responsible for managing the lifecycle of the downloaded artifacts.\n\n            (self.path contains the base path - hdfs:/some/path/run_id/artifacts)\n\n            :param artifact_path: Relative source path to the desired artifacts file or directory.\n            :param dst_path: Absolute path of the local filesystem destination directory to which\n                             to download the specified artifacts. This directory must already\n                             exist. If unspecified, the artifacts will be downloaded to a new,\n                             uniquely-named\n                             directory on the local filesystem.\n\n            :return: Absolute path of the local filesystem location containing the downloaded\n            artifacts - file/directory.\n        \"\"\"\n\n        hdfs_base_path = _resolve_base_path(self.path, artifact_path)\n        local_dir = _tmp_dir(dst_path)\n\n        with hdfs_system(host=self.host, port=self.port) as hdfs:\n\n            if not hdfs.isdir(hdfs_base_path):\n                local_path = os.path.join(local_dir, os.path.normpath(artifact_path))\n                _download_hdfs_file(hdfs, hdfs_base_path, local_path)\n                return local_path\n\n            for path, is_dir, _ in self._walk_path(hdfs, hdfs_base_path):\n\n                relative_path = _relative_path_remote(hdfs_base_path, path)\n                local_path = os.path.join(local_dir, relative_path) \\\n                    if relative_path else local_dir\n\n                if is_dir:\n                    mkdir(local_path)\n                else:\n                    _download_hdfs_file(hdfs, path, local_path)\n            return local_dir\n\n    def _download_file(self, remote_file_path, local_path):\n        raise MlflowException('This is not implemented. Should never be called.')\n\n\n@contextmanager\ndef hdfs_system(host, port):\n    \"\"\"\n        hdfs system context - Attempt to establish the connection to hdfs\n        and yields HadoopFileSystem\n\n    :param host: hostname or when relaying on the core-site.xml config use 'default'\n    :param port: port or when relaying on the core-site.xml config use 0\n    \"\"\"\n    import pyarrow as pa\n\n    driver = os.getenv('MLFLOW_HDFS_DRIVER') or 'libhdfs'\n    kerb_ticket = os.getenv('MLFLOW_KERBEROS_TICKET_CACHE')\n    kerberos_user = os.getenv('MLFLOW_KERBEROS_USER')\n    extra_conf = _parse_extra_conf(os.getenv('MLFLOW_PYARROW_EXTRA_CONF'))\n\n    connected = pa.hdfs.connect(host=host or 'default',\n                                port=port or 0,\n                                user=kerberos_user,\n                                driver=driver,\n                                kerb_ticket=kerb_ticket,\n                                extra_conf=extra_conf)\n    yield connected\n    connected.close()\n\n\ndef _resolve_connection_params(artifact_uri):\n    parsed = urllib.parse.urlparse(artifact_uri)\n    return parsed.hostname, parsed.port, parsed.path\n\n\ndef _resolve_base_path(path, artifact_path):\n    if path == artifact_path:\n        return path\n    if artifact_path:\n        return posixpath.join(path, artifact_path)\n    return path\n\n\ndef _relative_path(base_dir, subdir_path, path_module):\n    relative_path = path_module.relpath(subdir_path, base_dir)\n    return relative_path if relative_path is not '.' else None\n\n\ndef _relative_path_local(base_dir, subdir_path):\n    rel_path = _relative_path(base_dir, subdir_path, os.path)\n    return relative_path_to_artifact_path(rel_path) if rel_path is not None else None\n\n\ndef _relative_path_remote(base_dir, subdir_path):\n    return _relative_path(base_dir, subdir_path, posixpath)\n\n\ndef _tmp_dir(local_path):\n    return os.path.abspath(tempfile.mkdtemp(dir=local_path))\n\n\ndef _download_hdfs_file(hdfs, remote_file_path, local_file_path):\n    # Ensure all required directories exist. Without doing this nested files can't be downloaded.\n    dirs = os.path.dirname(local_file_path)\n    os.makedirs(dirs)\n    with open(local_file_path, 'wb') as f:\n        f.write(hdfs.open(remote_file_path, 'rb').read())\n\n\ndef _parse_extra_conf(extra_conf):\n    if extra_conf:\n        def as_pair(config):\n            key, val = config.split('=')\n            return key, val\n\n        list_of_key_val = [as_pair(conf) for conf in extra_conf.split(',')]\n        return dict(list_of_key_val)\n    return None\n", "path": "mlflow/store/hdfs_artifact_repo.py"}]}