We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
nrows= in dd.read_csv should fail loudly
We don't support this keyword argument.  We should fail loudly if someone tries to use it.


--- 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/dataframe/csv.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 from io import BytesIO
4 from warnings import warn
5 
6 try:
7     import psutil
8 except ImportError:
9     psutil = None
10 
11 import numpy as np
12 import pandas as pd
13 
14 from ..delayed import delayed
15 from .io import from_delayed
16 
17 from ..bytes import read_bytes
18 from ..bytes.compression import seekable_files, files as cfiles
19 
20 
21 delayed = delayed(pure=True)
22 
23 
24 def bytes_read_csv(b, header, kwargs, dtypes=None, columns=None,
25                    write_header=True, enforce=False):
26     """ Convert a block of bytes to a Pandas DataFrame
27 
28     Parameters
29     ----------
30     b: bytestring
31         The content to be parsed with pandas.read_csv
32     header: bytestring
33         An optional header to prepend to b
34     kwargs: dict
35         A dictionary of keyword arguments to be passed to pandas.read_csv
36     dtypes: dict
37         DTypes to assign to columns
38 
39     See Also:
40         dask.dataframe.csv.read_csv_from_bytes
41     """
42     bio = BytesIO()
43     if write_header and not b.startswith(header.rstrip()):
44         bio.write(header)
45     bio.write(b)
46     bio.seek(0)
47     df = pd.read_csv(bio, **kwargs)
48     if dtypes:
49         coerce_dtypes(df, dtypes)
50 
51     if enforce and columns and (list(df.columns) != list(columns)):
52         raise ValueError("Columns do not match", df.columns, columns)
53     return df
54 
55 
56 def coerce_dtypes(df, dtypes):
57     """ Coerce dataframe to dtypes safely
58 
59     Operates in place
60 
61     Parameters
62     ----------
63     df: Pandas DataFrame
64     dtypes: dict like {'x': float}
65     """
66     for c in df.columns:
67         if c in dtypes and df.dtypes[c] != dtypes[c]:
68             if (np.issubdtype(df.dtypes[c], np.floating) and
69                     np.issubdtype(dtypes[c], np.integer)):
70                 if (df[c] % 1).any():
71                     raise TypeError("Runtime type mismatch. "
72                     "Add {'%s': float} to dtype= keyword in read_csv" % c)
73             df[c] = df[c].astype(dtypes[c])
74 
75 
76 def read_csv_from_bytes(block_lists, header, head, kwargs, collection=True,
77                         enforce=False):
78     """ Convert blocks of bytes to a dask.dataframe or other high-level object
79 
80     This accepts a list of lists of values of bytes where each list corresponds
81     to one file, and the value of bytes concatenate to comprise the entire
82     file, in order.
83 
84     Parameters
85     ----------
86     block_lists: list of lists of delayed values of bytes
87         The lists of bytestrings where each list corresponds to one logical file
88     header: bytestring
89         The header, found at the front of the first file, to be prepended to
90         all blocks
91     head: pd.DataFrame
92         An example Pandas DataFrame to be used for metadata.
93         Can be ``None`` if ``collection==False``
94     kwargs: dict
95         Keyword arguments to pass down to ``pd.read_csv``
96     collection: boolean, optional (defaults to True)
97 
98     Returns
99     -------
100     A dask.dataframe or list of delayed values
101     """
102     dtypes = head.dtypes.to_dict()
103     columns = list(head.columns)
104     delayed_bytes_read_csv = delayed(bytes_read_csv)
105     dfs = []
106     for blocks in block_lists:
107         if not blocks:
108             continue
109         df = delayed_bytes_read_csv(blocks[0], header, kwargs, dtypes,
110                                     columns, write_header=False,
111                                     enforce=enforce)
112         dfs.append(df)
113         for b in blocks[1:]:
114             dfs.append(delayed_bytes_read_csv(b, header, kwargs, dtypes,
115                                               columns, enforce=enforce))
116 
117     if collection:
118         return from_delayed(dfs, head)
119     else:
120         return dfs
121 
122 
123 def auto_blocksize(total_memory, cpu_count):
124     memory_factor = 10
125     blocksize = int(total_memory // cpu_count / memory_factor)
126     return min(blocksize, int(64e6))
127 
128 
129 # guess blocksize if psutil is installed or use acceptable default one if not
130 if psutil is not None:
131     TOTAL_MEM = psutil.virtual_memory().total
132     CPU_COUNT = psutil.cpu_count()
133     AUTO_BLOCKSIZE = auto_blocksize(TOTAL_MEM, CPU_COUNT)
134 else:
135     AUTO_BLOCKSIZE = 2**25
136 
137 
138 def read_csv(urlpath, blocksize=AUTO_BLOCKSIZE, chunkbytes=None,
139         collection=True, lineterminator=None, compression=None,
140         sample=256000, enforce=False, storage_options=None, **kwargs):
141     """ Read CSV files into a Dask.DataFrame
142 
143     This parallelizes the ``pandas.read_csv`` file in the following ways:
144 
145     1.  It supports loading many files at once using globstrings as follows:
146 
147         >>> df = dd.read_csv('myfiles.*.csv')  # doctest: +SKIP
148 
149     2.  In some cases it can break up large files as follows:
150 
151         >>> df = dd.read_csv('largefile.csv', blocksize=25e6)  # 25MB chunks  # doctest: +SKIP
152 
153     3.  You can read CSV files from external resources (e.g. S3, HDFS)
154         providing a URL:
155 
156         >>> df = dd.read_csv('s3://bucket/myfiles.*.csv')  # doctest: +SKIP
157         >>> df = dd.read_csv('hdfs:///myfiles.*.csv')  # doctest: +SKIP
158         >>> df = dd.read_csv('hdfs://namenode.example.com/myfiles.*.csv')  # doctest: +SKIP
159 
160     Internally dd.read_csv uses pandas.read_csv and so supports many of the
161     same keyword arguments with the same performance guarantees.
162 
163     See the docstring for ``pandas.read_csv`` for more information on available
164     keyword arguments.
165 
166     Note that this function may fail if a CSV file includes quoted strings that
167     contain the line terminator.
168 
169     Parameters
170     ----------
171 
172     urlpath: string
173         Absolute or relative filepath, URL (may include protocols like
174         ``s3://``), or globstring for CSV files.
175     blocksize: int or None
176         Number of bytes by which to cut up larger files. Default value is
177         computed based on available physical memory and the number of cores.
178         If ``None``, use a single block for each file.
179     collection: boolean
180         Return a dask.dataframe if True or list of dask.delayed objects if False
181     sample: int
182         Number of bytes to use when determining dtypes
183     storage_options: dict
184         Extra options that make sense to a particular storage connection, e.g.
185         host, port, username, password, etc.
186     **kwargs: dict
187         Options to pass down to ``pandas.read_csv``
188     """
189     if lineterminator is not None and len(lineterminator) == 1:
190         kwargs['lineterminator'] = lineterminator
191     else:
192         lineterminator = '\n'
193     if chunkbytes is not None:
194         warn("Deprecation warning: chunksize csv keyword renamed to blocksize")
195         blocksize = chunkbytes
196     if 'index' in kwargs or 'index_col' in kwargs:
197         raise ValueError("Keyword 'index' not supported "
198                          "dd.read_csv(...).set_index('my-index') instead")
199     for kw in ['iterator', 'chunksize']:
200         if kw in kwargs:
201             raise ValueError("%s not supported for dd.read_csv" % kw)
202     if isinstance(kwargs.get('skiprows'), list):
203         raise TypeError("List of skiprows not supported for dd.read_csv")
204     if isinstance(kwargs.get('header'), list):
205         raise TypeError("List of header rows not supported for dd.read_csv")
206 
207     if blocksize and compression not in seekable_files:
208         warn("Warning %s compression does not support breaking apart files\n"
209              "Please ensure that each individual file can fit in memory and\n"
210              "use the keyword ``blocksize=None to remove this message``\n"
211              "Setting ``blocksize=None``" % compression)
212         blocksize = None
213     if compression not in seekable_files and compression not in cfiles:
214         raise NotImplementedError("Compression format %s not installed" %
215                                   compression)
216 
217     b_lineterminator = lineterminator.encode()
218     sample, values = read_bytes(urlpath, delimiter=b_lineterminator,
219                                          blocksize=blocksize,
220                                          sample=sample,
221                                          compression=compression,
222                                          **(storage_options or {}))
223 
224     if not isinstance(values[0], (tuple, list)):
225         values = [values]
226 
227     if 'nrows' in kwargs:
228         values = [[values[0][0]]]
229 
230     if kwargs.get('header', 'infer') is None:
231         header = b''
232     else:
233         header = sample.split(b_lineterminator)[0] + b_lineterminator
234 
235     head = pd.read_csv(BytesIO(sample), **kwargs)
236 
237     df = read_csv_from_bytes(values, header, head, kwargs,
238                              collection=collection, enforce=enforce)
239 
240     return df
241
```
--- 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/dataframe/csv.py b/dask/dataframe/csv.py
--- a/dask/dataframe/csv.py
+++ b/dask/dataframe/csv.py
@@ -199,6 +199,10 @@
     for kw in ['iterator', 'chunksize']:
         if kw in kwargs:
             raise ValueError("%s not supported for dd.read_csv" % kw)
+    if kwargs.get('nrows', None):
+        raise ValueError("The 'nrows' keyword is not supported by "
+                         "`dd.read_csv`. To achieve the same behavior, it's "
+                         "recommended to use `dd.read_csv(...).head(n=nrows)`")
     if isinstance(kwargs.get('skiprows'), list):
         raise TypeError("List of skiprows not supported for dd.read_csv")
     if isinstance(kwargs.get('header'), list):
@@ -224,9 +228,6 @@
     if not isinstance(values[0], (tuple, list)):
         values = [values]
 
-    if 'nrows' in kwargs:
-        values = [[values[0][0]]]
-
     if kwargs.get('header', 'infer') is None:
         header = b''
     else:

{"golden_diff": "diff --git a/dask/dataframe/csv.py b/dask/dataframe/csv.py\n--- a/dask/dataframe/csv.py\n+++ b/dask/dataframe/csv.py\n@@ -199,6 +199,10 @@\n     for kw in ['iterator', 'chunksize']:\n         if kw in kwargs:\n             raise ValueError(\"%s not supported for dd.read_csv\" % kw)\n+    if kwargs.get('nrows', None):\n+        raise ValueError(\"The 'nrows' keyword is not supported by \"\n+                         \"`dd.read_csv`. To achieve the same behavior, it's \"\n+                         \"recommended to use `dd.read_csv(...).head(n=nrows)`\")\n     if isinstance(kwargs.get('skiprows'), list):\n         raise TypeError(\"List of skiprows not supported for dd.read_csv\")\n     if isinstance(kwargs.get('header'), list):\n@@ -224,9 +228,6 @@\n     if not isinstance(values[0], (tuple, list)):\n         values = [values]\n \n-    if 'nrows' in kwargs:\n-        values = [[values[0][0]]]\n-\n     if kwargs.get('header', 'infer') is None:\n         header = b''\n     else:\n", "issue": "nrows= in dd.read_csv should fail loudly\nWe don't support this keyword argument.  We should fail loudly if someone tries to use it.\n\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom io import BytesIO\nfrom warnings import warn\n\ntry:\n    import psutil\nexcept ImportError:\n    psutil = None\n\nimport numpy as np\nimport pandas as pd\n\nfrom ..delayed import delayed\nfrom .io import from_delayed\n\nfrom ..bytes import read_bytes\nfrom ..bytes.compression import seekable_files, files as cfiles\n\n\ndelayed = delayed(pure=True)\n\n\ndef bytes_read_csv(b, header, kwargs, dtypes=None, columns=None,\n                   write_header=True, enforce=False):\n    \"\"\" Convert a block of bytes to a Pandas DataFrame\n\n    Parameters\n    ----------\n    b: bytestring\n        The content to be parsed with pandas.read_csv\n    header: bytestring\n        An optional header to prepend to b\n    kwargs: dict\n        A dictionary of keyword arguments to be passed to pandas.read_csv\n    dtypes: dict\n        DTypes to assign to columns\n\n    See Also:\n        dask.dataframe.csv.read_csv_from_bytes\n    \"\"\"\n    bio = BytesIO()\n    if write_header and not b.startswith(header.rstrip()):\n        bio.write(header)\n    bio.write(b)\n    bio.seek(0)\n    df = pd.read_csv(bio, **kwargs)\n    if dtypes:\n        coerce_dtypes(df, dtypes)\n\n    if enforce and columns and (list(df.columns) != list(columns)):\n        raise ValueError(\"Columns do not match\", df.columns, columns)\n    return df\n\n\ndef coerce_dtypes(df, dtypes):\n    \"\"\" Coerce dataframe to dtypes safely\n\n    Operates in place\n\n    Parameters\n    ----------\n    df: Pandas DataFrame\n    dtypes: dict like {'x': float}\n    \"\"\"\n    for c in df.columns:\n        if c in dtypes and df.dtypes[c] != dtypes[c]:\n            if (np.issubdtype(df.dtypes[c], np.floating) and\n                    np.issubdtype(dtypes[c], np.integer)):\n                if (df[c] % 1).any():\n                    raise TypeError(\"Runtime type mismatch. \"\n                    \"Add {'%s': float} to dtype= keyword in read_csv\" % c)\n            df[c] = df[c].astype(dtypes[c])\n\n\ndef read_csv_from_bytes(block_lists, header, head, kwargs, collection=True,\n                        enforce=False):\n    \"\"\" Convert blocks of bytes to a dask.dataframe or other high-level object\n\n    This accepts a list of lists of values of bytes where each list corresponds\n    to one file, and the value of bytes concatenate to comprise the entire\n    file, in order.\n\n    Parameters\n    ----------\n    block_lists: list of lists of delayed values of bytes\n        The lists of bytestrings where each list corresponds to one logical file\n    header: bytestring\n        The header, found at the front of the first file, to be prepended to\n        all blocks\n    head: pd.DataFrame\n        An example Pandas DataFrame to be used for metadata.\n        Can be ``None`` if ``collection==False``\n    kwargs: dict\n        Keyword arguments to pass down to ``pd.read_csv``\n    collection: boolean, optional (defaults to True)\n\n    Returns\n    -------\n    A dask.dataframe or list of delayed values\n    \"\"\"\n    dtypes = head.dtypes.to_dict()\n    columns = list(head.columns)\n    delayed_bytes_read_csv = delayed(bytes_read_csv)\n    dfs = []\n    for blocks in block_lists:\n        if not blocks:\n            continue\n        df = delayed_bytes_read_csv(blocks[0], header, kwargs, dtypes,\n                                    columns, write_header=False,\n                                    enforce=enforce)\n        dfs.append(df)\n        for b in blocks[1:]:\n            dfs.append(delayed_bytes_read_csv(b, header, kwargs, dtypes,\n                                              columns, enforce=enforce))\n\n    if collection:\n        return from_delayed(dfs, head)\n    else:\n        return dfs\n\n\ndef auto_blocksize(total_memory, cpu_count):\n    memory_factor = 10\n    blocksize = int(total_memory // cpu_count / memory_factor)\n    return min(blocksize, int(64e6))\n\n\n# guess blocksize if psutil is installed or use acceptable default one if not\nif psutil is not None:\n    TOTAL_MEM = psutil.virtual_memory().total\n    CPU_COUNT = psutil.cpu_count()\n    AUTO_BLOCKSIZE = auto_blocksize(TOTAL_MEM, CPU_COUNT)\nelse:\n    AUTO_BLOCKSIZE = 2**25\n\n\ndef read_csv(urlpath, blocksize=AUTO_BLOCKSIZE, chunkbytes=None,\n        collection=True, lineterminator=None, compression=None,\n        sample=256000, enforce=False, storage_options=None, **kwargs):\n    \"\"\" Read CSV files into a Dask.DataFrame\n\n    This parallelizes the ``pandas.read_csv`` file in the following ways:\n\n    1.  It supports loading many files at once using globstrings as follows:\n\n        >>> df = dd.read_csv('myfiles.*.csv')  # doctest: +SKIP\n\n    2.  In some cases it can break up large files as follows:\n\n        >>> df = dd.read_csv('largefile.csv', blocksize=25e6)  # 25MB chunks  # doctest: +SKIP\n\n    3.  You can read CSV files from external resources (e.g. S3, HDFS)\n        providing a URL:\n\n        >>> df = dd.read_csv('s3://bucket/myfiles.*.csv')  # doctest: +SKIP\n        >>> df = dd.read_csv('hdfs:///myfiles.*.csv')  # doctest: +SKIP\n        >>> df = dd.read_csv('hdfs://namenode.example.com/myfiles.*.csv')  # doctest: +SKIP\n\n    Internally dd.read_csv uses pandas.read_csv and so supports many of the\n    same keyword arguments with the same performance guarantees.\n\n    See the docstring for ``pandas.read_csv`` for more information on available\n    keyword arguments.\n\n    Note that this function may fail if a CSV file includes quoted strings that\n    contain the line terminator.\n\n    Parameters\n    ----------\n\n    urlpath: string\n        Absolute or relative filepath, URL (may include protocols like\n        ``s3://``), or globstring for CSV files.\n    blocksize: int or None\n        Number of bytes by which to cut up larger files. Default value is\n        computed based on available physical memory and the number of cores.\n        If ``None``, use a single block for each file.\n    collection: boolean\n        Return a dask.dataframe if True or list of dask.delayed objects if False\n    sample: int\n        Number of bytes to use when determining dtypes\n    storage_options: dict\n        Extra options that make sense to a particular storage connection, e.g.\n        host, port, username, password, etc.\n    **kwargs: dict\n        Options to pass down to ``pandas.read_csv``\n    \"\"\"\n    if lineterminator is not None and len(lineterminator) == 1:\n        kwargs['lineterminator'] = lineterminator\n    else:\n        lineterminator = '\\n'\n    if chunkbytes is not None:\n        warn(\"Deprecation warning: chunksize csv keyword renamed to blocksize\")\n        blocksize = chunkbytes\n    if 'index' in kwargs or 'index_col' in kwargs:\n        raise ValueError(\"Keyword 'index' not supported \"\n                         \"dd.read_csv(...).set_index('my-index') instead\")\n    for kw in ['iterator', 'chunksize']:\n        if kw in kwargs:\n            raise ValueError(\"%s not supported for dd.read_csv\" % kw)\n    if isinstance(kwargs.get('skiprows'), list):\n        raise TypeError(\"List of skiprows not supported for dd.read_csv\")\n    if isinstance(kwargs.get('header'), list):\n        raise TypeError(\"List of header rows not supported for dd.read_csv\")\n\n    if blocksize and compression not in seekable_files:\n        warn(\"Warning %s compression does not support breaking apart files\\n\"\n             \"Please ensure that each individual file can fit in memory and\\n\"\n             \"use the keyword ``blocksize=None to remove this message``\\n\"\n             \"Setting ``blocksize=None``\" % compression)\n        blocksize = None\n    if compression not in seekable_files and compression not in cfiles:\n        raise NotImplementedError(\"Compression format %s not installed\" %\n                                  compression)\n\n    b_lineterminator = lineterminator.encode()\n    sample, values = read_bytes(urlpath, delimiter=b_lineterminator,\n                                         blocksize=blocksize,\n                                         sample=sample,\n                                         compression=compression,\n                                         **(storage_options or {}))\n\n    if not isinstance(values[0], (tuple, list)):\n        values = [values]\n\n    if 'nrows' in kwargs:\n        values = [[values[0][0]]]\n\n    if kwargs.get('header', 'infer') is None:\n        header = b''\n    else:\n        header = sample.split(b_lineterminator)[0] + b_lineterminator\n\n    head = pd.read_csv(BytesIO(sample), **kwargs)\n\n    df = read_csv_from_bytes(values, header, head, kwargs,\n                             collection=collection, enforce=enforce)\n\n    return df\n", "path": "dask/dataframe/csv.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom io import BytesIO\nfrom warnings import warn\n\ntry:\n    import psutil\nexcept ImportError:\n    psutil = None\n\nimport numpy as np\nimport pandas as pd\n\nfrom ..delayed import delayed\nfrom .io import from_delayed\n\nfrom ..bytes import read_bytes\nfrom ..bytes.compression import seekable_files, files as cfiles\n\n\ndelayed = delayed(pure=True)\n\n\ndef bytes_read_csv(b, header, kwargs, dtypes=None, columns=None,\n                   write_header=True, enforce=False):\n    \"\"\" Convert a block of bytes to a Pandas DataFrame\n\n    Parameters\n    ----------\n    b: bytestring\n        The content to be parsed with pandas.read_csv\n    header: bytestring\n        An optional header to prepend to b\n    kwargs: dict\n        A dictionary of keyword arguments to be passed to pandas.read_csv\n    dtypes: dict\n        DTypes to assign to columns\n\n    See Also:\n        dask.dataframe.csv.read_csv_from_bytes\n    \"\"\"\n    bio = BytesIO()\n    if write_header and not b.startswith(header.rstrip()):\n        bio.write(header)\n    bio.write(b)\n    bio.seek(0)\n    df = pd.read_csv(bio, **kwargs)\n    if dtypes:\n        coerce_dtypes(df, dtypes)\n\n    if enforce and columns and (list(df.columns) != list(columns)):\n        raise ValueError(\"Columns do not match\", df.columns, columns)\n    return df\n\n\ndef coerce_dtypes(df, dtypes):\n    \"\"\" Coerce dataframe to dtypes safely\n\n    Operates in place\n\n    Parameters\n    ----------\n    df: Pandas DataFrame\n    dtypes: dict like {'x': float}\n    \"\"\"\n    for c in df.columns:\n        if c in dtypes and df.dtypes[c] != dtypes[c]:\n            if (np.issubdtype(df.dtypes[c], np.floating) and\n                    np.issubdtype(dtypes[c], np.integer)):\n                if (df[c] % 1).any():\n                    raise TypeError(\"Runtime type mismatch. \"\n                    \"Add {'%s': float} to dtype= keyword in read_csv\" % c)\n            df[c] = df[c].astype(dtypes[c])\n\n\ndef read_csv_from_bytes(block_lists, header, head, kwargs, collection=True,\n                        enforce=False):\n    \"\"\" Convert blocks of bytes to a dask.dataframe or other high-level object\n\n    This accepts a list of lists of values of bytes where each list corresponds\n    to one file, and the value of bytes concatenate to comprise the entire\n    file, in order.\n\n    Parameters\n    ----------\n    block_lists: list of lists of delayed values of bytes\n        The lists of bytestrings where each list corresponds to one logical file\n    header: bytestring\n        The header, found at the front of the first file, to be prepended to\n        all blocks\n    head: pd.DataFrame\n        An example Pandas DataFrame to be used for metadata.\n        Can be ``None`` if ``collection==False``\n    kwargs: dict\n        Keyword arguments to pass down to ``pd.read_csv``\n    collection: boolean, optional (defaults to True)\n\n    Returns\n    -------\n    A dask.dataframe or list of delayed values\n    \"\"\"\n    dtypes = head.dtypes.to_dict()\n    columns = list(head.columns)\n    delayed_bytes_read_csv = delayed(bytes_read_csv)\n    dfs = []\n    for blocks in block_lists:\n        if not blocks:\n            continue\n        df = delayed_bytes_read_csv(blocks[0], header, kwargs, dtypes,\n                                    columns, write_header=False,\n                                    enforce=enforce)\n        dfs.append(df)\n        for b in blocks[1:]:\n            dfs.append(delayed_bytes_read_csv(b, header, kwargs, dtypes,\n                                              columns, enforce=enforce))\n\n    if collection:\n        return from_delayed(dfs, head)\n    else:\n        return dfs\n\n\ndef auto_blocksize(total_memory, cpu_count):\n    memory_factor = 10\n    blocksize = int(total_memory // cpu_count / memory_factor)\n    return min(blocksize, int(64e6))\n\n\n# guess blocksize if psutil is installed or use acceptable default one if not\nif psutil is not None:\n    TOTAL_MEM = psutil.virtual_memory().total\n    CPU_COUNT = psutil.cpu_count()\n    AUTO_BLOCKSIZE = auto_blocksize(TOTAL_MEM, CPU_COUNT)\nelse:\n    AUTO_BLOCKSIZE = 2**25\n\n\ndef read_csv(urlpath, blocksize=AUTO_BLOCKSIZE, chunkbytes=None,\n        collection=True, lineterminator=None, compression=None,\n        sample=256000, enforce=False, storage_options=None, **kwargs):\n    \"\"\" Read CSV files into a Dask.DataFrame\n\n    This parallelizes the ``pandas.read_csv`` file in the following ways:\n\n    1.  It supports loading many files at once using globstrings as follows:\n\n        >>> df = dd.read_csv('myfiles.*.csv')  # doctest: +SKIP\n\n    2.  In some cases it can break up large files as follows:\n\n        >>> df = dd.read_csv('largefile.csv', blocksize=25e6)  # 25MB chunks  # doctest: +SKIP\n\n    3.  You can read CSV files from external resources (e.g. S3, HDFS)\n        providing a URL:\n\n        >>> df = dd.read_csv('s3://bucket/myfiles.*.csv')  # doctest: +SKIP\n        >>> df = dd.read_csv('hdfs:///myfiles.*.csv')  # doctest: +SKIP\n        >>> df = dd.read_csv('hdfs://namenode.example.com/myfiles.*.csv')  # doctest: +SKIP\n\n    Internally dd.read_csv uses pandas.read_csv and so supports many of the\n    same keyword arguments with the same performance guarantees.\n\n    See the docstring for ``pandas.read_csv`` for more information on available\n    keyword arguments.\n\n    Note that this function may fail if a CSV file includes quoted strings that\n    contain the line terminator.\n\n    Parameters\n    ----------\n\n    urlpath: string\n        Absolute or relative filepath, URL (may include protocols like\n        ``s3://``), or globstring for CSV files.\n    blocksize: int or None\n        Number of bytes by which to cut up larger files. Default value is\n        computed based on available physical memory and the number of cores.\n        If ``None``, use a single block for each file.\n    collection: boolean\n        Return a dask.dataframe if True or list of dask.delayed objects if False\n    sample: int\n        Number of bytes to use when determining dtypes\n    storage_options: dict\n        Extra options that make sense to a particular storage connection, e.g.\n        host, port, username, password, etc.\n    **kwargs: dict\n        Options to pass down to ``pandas.read_csv``\n    \"\"\"\n    if lineterminator is not None and len(lineterminator) == 1:\n        kwargs['lineterminator'] = lineterminator\n    else:\n        lineterminator = '\\n'\n    if chunkbytes is not None:\n        warn(\"Deprecation warning: chunksize csv keyword renamed to blocksize\")\n        blocksize = chunkbytes\n    if 'index' in kwargs or 'index_col' in kwargs:\n        raise ValueError(\"Keyword 'index' not supported \"\n                         \"dd.read_csv(...).set_index('my-index') instead\")\n    for kw in ['iterator', 'chunksize']:\n        if kw in kwargs:\n            raise ValueError(\"%s not supported for dd.read_csv\" % kw)\n    if kwargs.get('nrows', None):\n        raise ValueError(\"The 'nrows' keyword is not supported by \"\n                         \"`dd.read_csv`. To achieve the same behavior, it's \"\n                         \"recommended to use `dd.read_csv(...).head(n=nrows)`\")\n    if isinstance(kwargs.get('skiprows'), list):\n        raise TypeError(\"List of skiprows not supported for dd.read_csv\")\n    if isinstance(kwargs.get('header'), list):\n        raise TypeError(\"List of header rows not supported for dd.read_csv\")\n\n    if blocksize and compression not in seekable_files:\n        warn(\"Warning %s compression does not support breaking apart files\\n\"\n             \"Please ensure that each individual file can fit in memory and\\n\"\n             \"use the keyword ``blocksize=None to remove this message``\\n\"\n             \"Setting ``blocksize=None``\" % compression)\n        blocksize = None\n    if compression not in seekable_files and compression not in cfiles:\n        raise NotImplementedError(\"Compression format %s not installed\" %\n                                  compression)\n\n    b_lineterminator = lineterminator.encode()\n    sample, values = read_bytes(urlpath, delimiter=b_lineterminator,\n                                         blocksize=blocksize,\n                                         sample=sample,\n                                         compression=compression,\n                                         **(storage_options or {}))\n\n    if not isinstance(values[0], (tuple, list)):\n        values = [values]\n\n    if kwargs.get('header', 'infer') is None:\n        header = b''\n    else:\n        header = sample.split(b_lineterminator)[0] + b_lineterminator\n\n    head = pd.read_csv(BytesIO(sample), **kwargs)\n\n    df = read_csv_from_bytes(values, header, head, kwargs,\n                             collection=collection, enforce=enforce)\n\n    return df\n", "path": "dask/dataframe/csv.py"}]}