We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
NotImplementedError: Unknown protocol hdfs
After running into issue #1232, I upgraded to the the source version.  hdfs3 is installed and working, but

```
import hdfs3
from dask import dataframe as dd

namenode = 'mynode.com'
port = 8020
hdfs = hdfs3.HDFileSystem(namenode, port)
hdfs.list('/path/to/file.csv')
# returns file node information
data = df.read_csv("hdfs://{0}/path/to/file.csv".format(namenode))
```

This produces the following error

```
NotImplementedErrorTraceback (most recent call last)
<ipython-input-30-2d58fa3887f0> in <module>()
----> 1 data = dd.read_csv('hdfs://{0}/path.to/file.csv'.format(namenode))

/usr/local/share/virtualenvs/cartman/lib/python2.7/site-packages/dask/dataframe/csv.pyc in    read_csv(filename, blocksize, chunkbytes, collection, lineterminator, compression, sample, enforce,  storage_options, **kwargs)

/usr/local/share/virtualenvs/cartman/lib/python2.7/site-packages/dask/bytes/core.pyc in read_bytes(path,   delimiter, not_zero, blocksize, sample, compression, **kwargs)

NotImplementedError: Unknown protocol hdfs://mynode.com/path/to/file.csv
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `dask/bytes/core.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 import io
4 
5 from toolz import merge
6 
7 from .compression import seekable_files, files as compress_files
8 from .utils import SeekableFile
9 from ..compatibility import PY2
10 from ..delayed import delayed
11 from ..utils import infer_storage_options, system_encoding
12 
13 delayed = delayed(pure=True)
14 
15 # Global registration dictionaries for backend storage functions
16 # See docstrings to functions below for more information
17 _read_bytes = dict()
18 _open_files = dict()
19 _open_text_files = dict()
20 
21 
22 def read_bytes(urlpath, delimiter=None, not_zero=False, blocksize=2**27,
23                      sample=True, compression=None, **kwargs):
24     """ Convert path to a list of delayed values
25 
26     The path may be a filename like ``'2015-01-01.csv'`` or a globstring
27     like ``'2015-*-*.csv'``.
28 
29     The path may be preceded by a protocol, like ``s3://`` or ``hdfs://`` if
30     those libraries are installed.
31 
32     This cleanly breaks data by a delimiter if given, so that block boundaries
33     start directly after a delimiter and end on the delimiter.
34 
35     Parameters
36     ----------
37     urlpath: string
38         Absolute or relative filepath, URL (may include protocols like
39         ``s3://``), or globstring pointing to data.
40     delimiter: bytes
41         An optional delimiter, like ``b'\n'`` on which to split blocks of bytes
42     not_zero: force seek of start-of-file delimiter, discarding header
43     blocksize: int (=128MB)
44         Chunk size
45     compression: string or None
46         String like 'gzip' or 'xz'.  Must support efficient random access.
47     sample: bool, int
48         Whether or not to return a sample from the first 10k bytes
49     **kwargs: dict
50         Extra options that make sense to a particular storage connection, e.g.
51         host, port, username, password, etc.
52 
53     Examples
54     --------
55     >>> sample, blocks = read_bytes('2015-*-*.csv', delimiter=b'\\n')  # doctest: +SKIP
56     >>> sample, blocks = read_bytes('s3://bucket/2015-*-*.csv', delimiter=b'\\n')  # doctest: +SKIP
57 
58     Returns
59     -------
60     10kB sample header and list of ``dask.Delayed`` objects or list of lists of
61     delayed objects if ``fn`` is a globstring.
62     """
63     if compression is not None and compression not in compress_files:
64         raise ValueError("Compression type %s not supported" % compression)
65 
66     storage_options = infer_storage_options(urlpath,
67                                             inherit_storage_options=kwargs)
68     protocol = storage_options.pop('protocol')
69     try:
70         read_bytes = _read_bytes[protocol]
71     except KeyError:
72         raise NotImplementedError("Unknown protocol %s (%s)" %
73                                   (protocol, urlpath))
74 
75     return read_bytes(storage_options.pop('path'), delimiter=delimiter,
76             not_zero=not_zero, blocksize=blocksize, sample=sample,
77             compression=compression, **storage_options)
78 
79 
80 def open_files_by(open_files_backend, path, compression=None, **kwargs):
81     """ Given open files backend and path return dask.delayed file-like objects
82 
83     NOTE: This is an internal helper function, please refer to
84     :func:`open_files` documentation for more details.
85 
86     Parameters
87     ----------
88     path: string
89         Filepath or globstring
90     compression: string
91         Compression to use.  See ``dask.bytes.compression.files`` for options.
92     **kwargs: dict
93         Extra options that make sense to a particular storage connection, e.g.
94         host, port, username, password, etc.
95 
96     Returns
97     -------
98     List of ``dask.delayed`` objects that compute to file-like objects
99     """
100     files = open_files_backend(path, **kwargs)
101 
102     if compression:
103         decompress = merge(seekable_files, compress_files)[compression]
104         if PY2:
105             files = [delayed(SeekableFile)(file) for file in files]
106         files = [delayed(decompress)(file) for file in files]
107 
108     return files
109 
110 
111 def open_files(urlpath, compression=None, **kwargs):
112     """ Given path return dask.delayed file-like objects
113 
114     Parameters
115     ----------
116     urlpath: string
117         Absolute or relative filepath, URL (may include protocols like
118         ``s3://``), or globstring pointing to data.
119     compression: string
120         Compression to use.  See ``dask.bytes.compression.files`` for options.
121     **kwargs: dict
122         Extra options that make sense to a particular storage connection, e.g.
123         host, port, username, password, etc.
124 
125     Examples
126     --------
127     >>> files = open_files('2015-*-*.csv')  # doctest: +SKIP
128     >>> files = open_files('s3://bucket/2015-*-*.csv.gz', compression='gzip')  # doctest: +SKIP
129 
130     Returns
131     -------
132     List of ``dask.delayed`` objects that compute to file-like objects
133     """
134     if compression is not None and compression not in compress_files:
135         raise ValueError("Compression type %s not supported" % compression)
136 
137     storage_options = infer_storage_options(urlpath,
138                                             inherit_storage_options=kwargs)
139     protocol = storage_options.pop('protocol')
140     try:
141         open_files_backend = _open_files[protocol]
142     except KeyError:
143         raise NotImplementedError("Unknown protocol %s (%s)" %
144                                   (protocol, urlpath))
145 
146     return open_files_by(open_files_backend, storage_options.pop('path'),
147                          compression=compression, **storage_options)
148 
149 
150 def open_text_files(urlpath, encoding=system_encoding, errors='strict',
151         compression=None, **kwargs):
152     """ Given path return dask.delayed file-like objects in text mode
153 
154     Parameters
155     ----------
156     urlpath: string
157         Absolute or relative filepath, URL (may include protocols like
158         ``s3://``), or globstring pointing to data.
159     encoding: string
160     errors: string
161     compression: string
162         Compression to use.  See ``dask.bytes.compression.files`` for options.
163     **kwargs: dict
164         Extra options that make sense to a particular storage connection, e.g.
165         host, port, username, password, etc.
166 
167     Examples
168     --------
169     >>> files = open_text_files('2015-*-*.csv', encoding='utf-8')  # doctest: +SKIP
170     >>> files = open_text_files('s3://bucket/2015-*-*.csv')  # doctest: +SKIP
171 
172     Returns
173     -------
174     List of ``dask.delayed`` objects that compute to text file-like objects
175     """
176     if compression is not None and compression not in compress_files:
177         raise ValueError("Compression type %s not supported" % compression)
178 
179     storage_options = infer_storage_options(urlpath,
180                                             inherit_storage_options=kwargs)
181     path = storage_options.pop('path')
182     protocol = storage_options.pop('protocol')
183     if protocol in _open_text_files and compression is None:
184         return _open_text_files[protocol](path,
185                                           encoding=encoding,
186                                           errors=errors,
187                                           **storage_options)
188     elif protocol in _open_files:
189         files = open_files_by(_open_files[protocol],
190                               path,
191                               compression=compression,
192                               **storage_options)
193         if PY2:
194             files = [delayed(SeekableFile)(file) for file in files]
195         return [delayed(io.TextIOWrapper)(file, encoding=encoding,
196                                           errors=errors) for file in files]
197     else:
198         raise NotImplementedError("Unknown protocol %s (%s)" %
199                                   (protocol, urlpath))
200
```
--- 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/dask/bytes/core.py b/dask/bytes/core.py
--- a/dask/bytes/core.py
+++ b/dask/bytes/core.py
@@ -66,6 +66,7 @@
     storage_options = infer_storage_options(urlpath,
                                             inherit_storage_options=kwargs)
     protocol = storage_options.pop('protocol')
+    ensure_protocol(protocol)
     try:
         read_bytes = _read_bytes[protocol]
     except KeyError:
@@ -137,6 +138,7 @@
     storage_options = infer_storage_options(urlpath,
                                             inherit_storage_options=kwargs)
     protocol = storage_options.pop('protocol')
+    ensure_protocol(protocol)
     try:
         open_files_backend = _open_files[protocol]
     except KeyError:
@@ -180,6 +182,7 @@
                                             inherit_storage_options=kwargs)
     path = storage_options.pop('path')
     protocol = storage_options.pop('protocol')
+    ensure_protocol(protocol)
     if protocol in _open_text_files and compression is None:
         return _open_text_files[protocol](path,
                                           encoding=encoding,
@@ -197,3 +200,28 @@
     else:
         raise NotImplementedError("Unknown protocol %s (%s)" %
                                   (protocol, urlpath))
+
+
+def ensure_protocol(protocol):
+    if protocol in _read_bytes or protocol in _open_files:
+        return
+
+    if protocol == 's3':
+        try:
+            import dask.s3
+        except ImportError:
+            raise ImportError("Need to install `s3fs` library for s3 support\n"
+                    "    conda install s3fs -c conda-forge\n"
+                    "    or\n"
+                    "    pip install s3fs")
+
+    elif protocol == 'hdfs':
+        try:
+            import distributed.hdfs
+        except ImportError:
+            raise ImportError("Need to install `distributed` and `hdfs3` "
+                    "for HDFS support\n"
+                    "    conda install distributed hdfs3 -c conda-forge")
+
+    else:
+        raise ValueError("Unknown protocol %s" % protocol)

{"golden_diff": "diff --git a/dask/bytes/core.py b/dask/bytes/core.py\n--- a/dask/bytes/core.py\n+++ b/dask/bytes/core.py\n@@ -66,6 +66,7 @@\n     storage_options = infer_storage_options(urlpath,\n                                             inherit_storage_options=kwargs)\n     protocol = storage_options.pop('protocol')\n+    ensure_protocol(protocol)\n     try:\n         read_bytes = _read_bytes[protocol]\n     except KeyError:\n@@ -137,6 +138,7 @@\n     storage_options = infer_storage_options(urlpath,\n                                             inherit_storage_options=kwargs)\n     protocol = storage_options.pop('protocol')\n+    ensure_protocol(protocol)\n     try:\n         open_files_backend = _open_files[protocol]\n     except KeyError:\n@@ -180,6 +182,7 @@\n                                             inherit_storage_options=kwargs)\n     path = storage_options.pop('path')\n     protocol = storage_options.pop('protocol')\n+    ensure_protocol(protocol)\n     if protocol in _open_text_files and compression is None:\n         return _open_text_files[protocol](path,\n                                           encoding=encoding,\n@@ -197,3 +200,28 @@\n     else:\n         raise NotImplementedError(\"Unknown protocol %s (%s)\" %\n                                   (protocol, urlpath))\n+\n+\n+def ensure_protocol(protocol):\n+    if protocol in _read_bytes or protocol in _open_files:\n+        return\n+\n+    if protocol == 's3':\n+        try:\n+            import dask.s3\n+        except ImportError:\n+            raise ImportError(\"Need to install `s3fs` library for s3 support\\n\"\n+                    \"    conda install s3fs -c conda-forge\\n\"\n+                    \"    or\\n\"\n+                    \"    pip install s3fs\")\n+\n+    elif protocol == 'hdfs':\n+        try:\n+            import distributed.hdfs\n+        except ImportError:\n+            raise ImportError(\"Need to install `distributed` and `hdfs3` \"\n+                    \"for HDFS support\\n\"\n+                    \"    conda install distributed hdfs3 -c conda-forge\")\n+\n+    else:\n+        raise ValueError(\"Unknown protocol %s\" % protocol)\n", "issue": "NotImplementedError: Unknown protocol hdfs\nAfter running into issue #1232, I upgraded to the the source version.  hdfs3 is installed and working, but\n\n```\nimport hdfs3\nfrom dask import dataframe as dd\n\nnamenode = 'mynode.com'\nport = 8020\nhdfs = hdfs3.HDFileSystem(namenode, port)\nhdfs.list('/path/to/file.csv')\n# returns file node information\ndata = df.read_csv(\"hdfs://{0}/path/to/file.csv\".format(namenode))\n```\n\nThis produces the following error\n\n```\nNotImplementedErrorTraceback (most recent call last)\n<ipython-input-30-2d58fa3887f0> in <module>()\n----> 1 data = dd.read_csv('hdfs://{0}/path.to/file.csv'.format(namenode))\n\n/usr/local/share/virtualenvs/cartman/lib/python2.7/site-packages/dask/dataframe/csv.pyc in    read_csv(filename, blocksize, chunkbytes, collection, lineterminator, compression, sample, enforce,  storage_options, **kwargs)\n\n/usr/local/share/virtualenvs/cartman/lib/python2.7/site-packages/dask/bytes/core.pyc in read_bytes(path,   delimiter, not_zero, blocksize, sample, compression, **kwargs)\n\nNotImplementedError: Unknown protocol hdfs://mynode.com/path/to/file.csv\n```\n\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nimport io\n\nfrom toolz import merge\n\nfrom .compression import seekable_files, files as compress_files\nfrom .utils import SeekableFile\nfrom ..compatibility import PY2\nfrom ..delayed import delayed\nfrom ..utils import infer_storage_options, system_encoding\n\ndelayed = delayed(pure=True)\n\n# Global registration dictionaries for backend storage functions\n# See docstrings to functions below for more information\n_read_bytes = dict()\n_open_files = dict()\n_open_text_files = dict()\n\n\ndef read_bytes(urlpath, delimiter=None, not_zero=False, blocksize=2**27,\n                     sample=True, compression=None, **kwargs):\n    \"\"\" Convert path to a list of delayed values\n\n    The path may be a filename like ``'2015-01-01.csv'`` or a globstring\n    like ``'2015-*-*.csv'``.\n\n    The path may be preceded by a protocol, like ``s3://`` or ``hdfs://`` if\n    those libraries are installed.\n\n    This cleanly breaks data by a delimiter if given, so that block boundaries\n    start directly after a delimiter and end on the delimiter.\n\n    Parameters\n    ----------\n    urlpath: string\n        Absolute or relative filepath, URL (may include protocols like\n        ``s3://``), or globstring pointing to data.\n    delimiter: bytes\n        An optional delimiter, like ``b'\\n'`` on which to split blocks of bytes\n    not_zero: force seek of start-of-file delimiter, discarding header\n    blocksize: int (=128MB)\n        Chunk size\n    compression: string or None\n        String like 'gzip' or 'xz'.  Must support efficient random access.\n    sample: bool, int\n        Whether or not to return a sample from the first 10k bytes\n    **kwargs: dict\n        Extra options that make sense to a particular storage connection, e.g.\n        host, port, username, password, etc.\n\n    Examples\n    --------\n    >>> sample, blocks = read_bytes('2015-*-*.csv', delimiter=b'\\\\n')  # doctest: +SKIP\n    >>> sample, blocks = read_bytes('s3://bucket/2015-*-*.csv', delimiter=b'\\\\n')  # doctest: +SKIP\n\n    Returns\n    -------\n    10kB sample header and list of ``dask.Delayed`` objects or list of lists of\n    delayed objects if ``fn`` is a globstring.\n    \"\"\"\n    if compression is not None and compression not in compress_files:\n        raise ValueError(\"Compression type %s not supported\" % compression)\n\n    storage_options = infer_storage_options(urlpath,\n                                            inherit_storage_options=kwargs)\n    protocol = storage_options.pop('protocol')\n    try:\n        read_bytes = _read_bytes[protocol]\n    except KeyError:\n        raise NotImplementedError(\"Unknown protocol %s (%s)\" %\n                                  (protocol, urlpath))\n\n    return read_bytes(storage_options.pop('path'), delimiter=delimiter,\n            not_zero=not_zero, blocksize=blocksize, sample=sample,\n            compression=compression, **storage_options)\n\n\ndef open_files_by(open_files_backend, path, compression=None, **kwargs):\n    \"\"\" Given open files backend and path return dask.delayed file-like objects\n\n    NOTE: This is an internal helper function, please refer to\n    :func:`open_files` documentation for more details.\n\n    Parameters\n    ----------\n    path: string\n        Filepath or globstring\n    compression: string\n        Compression to use.  See ``dask.bytes.compression.files`` for options.\n    **kwargs: dict\n        Extra options that make sense to a particular storage connection, e.g.\n        host, port, username, password, etc.\n\n    Returns\n    -------\n    List of ``dask.delayed`` objects that compute to file-like objects\n    \"\"\"\n    files = open_files_backend(path, **kwargs)\n\n    if compression:\n        decompress = merge(seekable_files, compress_files)[compression]\n        if PY2:\n            files = [delayed(SeekableFile)(file) for file in files]\n        files = [delayed(decompress)(file) for file in files]\n\n    return files\n\n\ndef open_files(urlpath, compression=None, **kwargs):\n    \"\"\" Given path return dask.delayed file-like objects\n\n    Parameters\n    ----------\n    urlpath: string\n        Absolute or relative filepath, URL (may include protocols like\n        ``s3://``), or globstring pointing to data.\n    compression: string\n        Compression to use.  See ``dask.bytes.compression.files`` for options.\n    **kwargs: dict\n        Extra options that make sense to a particular storage connection, e.g.\n        host, port, username, password, etc.\n\n    Examples\n    --------\n    >>> files = open_files('2015-*-*.csv')  # doctest: +SKIP\n    >>> files = open_files('s3://bucket/2015-*-*.csv.gz', compression='gzip')  # doctest: +SKIP\n\n    Returns\n    -------\n    List of ``dask.delayed`` objects that compute to file-like objects\n    \"\"\"\n    if compression is not None and compression not in compress_files:\n        raise ValueError(\"Compression type %s not supported\" % compression)\n\n    storage_options = infer_storage_options(urlpath,\n                                            inherit_storage_options=kwargs)\n    protocol = storage_options.pop('protocol')\n    try:\n        open_files_backend = _open_files[protocol]\n    except KeyError:\n        raise NotImplementedError(\"Unknown protocol %s (%s)\" %\n                                  (protocol, urlpath))\n\n    return open_files_by(open_files_backend, storage_options.pop('path'),\n                         compression=compression, **storage_options)\n\n\ndef open_text_files(urlpath, encoding=system_encoding, errors='strict',\n        compression=None, **kwargs):\n    \"\"\" Given path return dask.delayed file-like objects in text mode\n\n    Parameters\n    ----------\n    urlpath: string\n        Absolute or relative filepath, URL (may include protocols like\n        ``s3://``), or globstring pointing to data.\n    encoding: string\n    errors: string\n    compression: string\n        Compression to use.  See ``dask.bytes.compression.files`` for options.\n    **kwargs: dict\n        Extra options that make sense to a particular storage connection, e.g.\n        host, port, username, password, etc.\n\n    Examples\n    --------\n    >>> files = open_text_files('2015-*-*.csv', encoding='utf-8')  # doctest: +SKIP\n    >>> files = open_text_files('s3://bucket/2015-*-*.csv')  # doctest: +SKIP\n\n    Returns\n    -------\n    List of ``dask.delayed`` objects that compute to text file-like objects\n    \"\"\"\n    if compression is not None and compression not in compress_files:\n        raise ValueError(\"Compression type %s not supported\" % compression)\n\n    storage_options = infer_storage_options(urlpath,\n                                            inherit_storage_options=kwargs)\n    path = storage_options.pop('path')\n    protocol = storage_options.pop('protocol')\n    if protocol in _open_text_files and compression is None:\n        return _open_text_files[protocol](path,\n                                          encoding=encoding,\n                                          errors=errors,\n                                          **storage_options)\n    elif protocol in _open_files:\n        files = open_files_by(_open_files[protocol],\n                              path,\n                              compression=compression,\n                              **storage_options)\n        if PY2:\n            files = [delayed(SeekableFile)(file) for file in files]\n        return [delayed(io.TextIOWrapper)(file, encoding=encoding,\n                                          errors=errors) for file in files]\n    else:\n        raise NotImplementedError(\"Unknown protocol %s (%s)\" %\n                                  (protocol, urlpath))\n", "path": "dask/bytes/core.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nimport io\n\nfrom toolz import merge\n\nfrom .compression import seekable_files, files as compress_files\nfrom .utils import SeekableFile\nfrom ..compatibility import PY2\nfrom ..delayed import delayed\nfrom ..utils import infer_storage_options, system_encoding\n\ndelayed = delayed(pure=True)\n\n# Global registration dictionaries for backend storage functions\n# See docstrings to functions below for more information\n_read_bytes = dict()\n_open_files = dict()\n_open_text_files = dict()\n\n\ndef read_bytes(urlpath, delimiter=None, not_zero=False, blocksize=2**27,\n                     sample=True, compression=None, **kwargs):\n    \"\"\" Convert path to a list of delayed values\n\n    The path may be a filename like ``'2015-01-01.csv'`` or a globstring\n    like ``'2015-*-*.csv'``.\n\n    The path may be preceded by a protocol, like ``s3://`` or ``hdfs://`` if\n    those libraries are installed.\n\n    This cleanly breaks data by a delimiter if given, so that block boundaries\n    start directly after a delimiter and end on the delimiter.\n\n    Parameters\n    ----------\n    urlpath: string\n        Absolute or relative filepath, URL (may include protocols like\n        ``s3://``), or globstring pointing to data.\n    delimiter: bytes\n        An optional delimiter, like ``b'\\n'`` on which to split blocks of bytes\n    not_zero: force seek of start-of-file delimiter, discarding header\n    blocksize: int (=128MB)\n        Chunk size\n    compression: string or None\n        String like 'gzip' or 'xz'.  Must support efficient random access.\n    sample: bool, int\n        Whether or not to return a sample from the first 10k bytes\n    **kwargs: dict\n        Extra options that make sense to a particular storage connection, e.g.\n        host, port, username, password, etc.\n\n    Examples\n    --------\n    >>> sample, blocks = read_bytes('2015-*-*.csv', delimiter=b'\\\\n')  # doctest: +SKIP\n    >>> sample, blocks = read_bytes('s3://bucket/2015-*-*.csv', delimiter=b'\\\\n')  # doctest: +SKIP\n\n    Returns\n    -------\n    10kB sample header and list of ``dask.Delayed`` objects or list of lists of\n    delayed objects if ``fn`` is a globstring.\n    \"\"\"\n    if compression is not None and compression not in compress_files:\n        raise ValueError(\"Compression type %s not supported\" % compression)\n\n    storage_options = infer_storage_options(urlpath,\n                                            inherit_storage_options=kwargs)\n    protocol = storage_options.pop('protocol')\n    ensure_protocol(protocol)\n    try:\n        read_bytes = _read_bytes[protocol]\n    except KeyError:\n        raise NotImplementedError(\"Unknown protocol %s (%s)\" %\n                                  (protocol, urlpath))\n\n    return read_bytes(storage_options.pop('path'), delimiter=delimiter,\n            not_zero=not_zero, blocksize=blocksize, sample=sample,\n            compression=compression, **storage_options)\n\n\ndef open_files_by(open_files_backend, path, compression=None, **kwargs):\n    \"\"\" Given open files backend and path return dask.delayed file-like objects\n\n    NOTE: This is an internal helper function, please refer to\n    :func:`open_files` documentation for more details.\n\n    Parameters\n    ----------\n    path: string\n        Filepath or globstring\n    compression: string\n        Compression to use.  See ``dask.bytes.compression.files`` for options.\n    **kwargs: dict\n        Extra options that make sense to a particular storage connection, e.g.\n        host, port, username, password, etc.\n\n    Returns\n    -------\n    List of ``dask.delayed`` objects that compute to file-like objects\n    \"\"\"\n    files = open_files_backend(path, **kwargs)\n\n    if compression:\n        decompress = merge(seekable_files, compress_files)[compression]\n        if PY2:\n            files = [delayed(SeekableFile)(file) for file in files]\n        files = [delayed(decompress)(file) for file in files]\n\n    return files\n\n\ndef open_files(urlpath, compression=None, **kwargs):\n    \"\"\" Given path return dask.delayed file-like objects\n\n    Parameters\n    ----------\n    urlpath: string\n        Absolute or relative filepath, URL (may include protocols like\n        ``s3://``), or globstring pointing to data.\n    compression: string\n        Compression to use.  See ``dask.bytes.compression.files`` for options.\n    **kwargs: dict\n        Extra options that make sense to a particular storage connection, e.g.\n        host, port, username, password, etc.\n\n    Examples\n    --------\n    >>> files = open_files('2015-*-*.csv')  # doctest: +SKIP\n    >>> files = open_files('s3://bucket/2015-*-*.csv.gz', compression='gzip')  # doctest: +SKIP\n\n    Returns\n    -------\n    List of ``dask.delayed`` objects that compute to file-like objects\n    \"\"\"\n    if compression is not None and compression not in compress_files:\n        raise ValueError(\"Compression type %s not supported\" % compression)\n\n    storage_options = infer_storage_options(urlpath,\n                                            inherit_storage_options=kwargs)\n    protocol = storage_options.pop('protocol')\n    ensure_protocol(protocol)\n    try:\n        open_files_backend = _open_files[protocol]\n    except KeyError:\n        raise NotImplementedError(\"Unknown protocol %s (%s)\" %\n                                  (protocol, urlpath))\n\n    return open_files_by(open_files_backend, storage_options.pop('path'),\n                         compression=compression, **storage_options)\n\n\ndef open_text_files(urlpath, encoding=system_encoding, errors='strict',\n        compression=None, **kwargs):\n    \"\"\" Given path return dask.delayed file-like objects in text mode\n\n    Parameters\n    ----------\n    urlpath: string\n        Absolute or relative filepath, URL (may include protocols like\n        ``s3://``), or globstring pointing to data.\n    encoding: string\n    errors: string\n    compression: string\n        Compression to use.  See ``dask.bytes.compression.files`` for options.\n    **kwargs: dict\n        Extra options that make sense to a particular storage connection, e.g.\n        host, port, username, password, etc.\n\n    Examples\n    --------\n    >>> files = open_text_files('2015-*-*.csv', encoding='utf-8')  # doctest: +SKIP\n    >>> files = open_text_files('s3://bucket/2015-*-*.csv')  # doctest: +SKIP\n\n    Returns\n    -------\n    List of ``dask.delayed`` objects that compute to text file-like objects\n    \"\"\"\n    if compression is not None and compression not in compress_files:\n        raise ValueError(\"Compression type %s not supported\" % compression)\n\n    storage_options = infer_storage_options(urlpath,\n                                            inherit_storage_options=kwargs)\n    path = storage_options.pop('path')\n    protocol = storage_options.pop('protocol')\n    ensure_protocol(protocol)\n    if protocol in _open_text_files and compression is None:\n        return _open_text_files[protocol](path,\n                                          encoding=encoding,\n                                          errors=errors,\n                                          **storage_options)\n    elif protocol in _open_files:\n        files = open_files_by(_open_files[protocol],\n                              path,\n                              compression=compression,\n                              **storage_options)\n        if PY2:\n            files = [delayed(SeekableFile)(file) for file in files]\n        return [delayed(io.TextIOWrapper)(file, encoding=encoding,\n                                          errors=errors) for file in files]\n    else:\n        raise NotImplementedError(\"Unknown protocol %s (%s)\" %\n                                  (protocol, urlpath))\n\n\ndef ensure_protocol(protocol):\n    if protocol in _read_bytes or protocol in _open_files:\n        return\n\n    if protocol == 's3':\n        try:\n            import dask.s3\n        except ImportError:\n            raise ImportError(\"Need to install `s3fs` library for s3 support\\n\"\n                    \"    conda install s3fs -c conda-forge\\n\"\n                    \"    or\\n\"\n                    \"    pip install s3fs\")\n\n    elif protocol == 'hdfs':\n        try:\n            import distributed.hdfs\n        except ImportError:\n            raise ImportError(\"Need to install `distributed` and `hdfs3` \"\n                    \"for HDFS support\\n\"\n                    \"    conda install distributed hdfs3 -c conda-forge\")\n\n    else:\n        raise ValueError(\"Unknown protocol %s\" % protocol)\n", "path": "dask/bytes/core.py"}]}