Repository: dask/dask
Pull request: 4809 fix extra cr
- [x] Tests added / passed
- [x] Passes `flake8 dask`

Fixes #4809.

Includes the following new test, which demonstrates the issue, and did, indeed, fail prior to the fix here.

```python
def test_to_csv_line_ending():
    df = pd.DataFrame({'x': [0]})
    ddf = dd.from_pandas(df, npartitions=1)
    expected = {b'0\r\n', b'0\n'} # either/or
    # For comparison...
    # unexpected = {b'0\r\r\n'}
    # This test addresses GH4809, and checks that only (at most) one
    #  '\r' character is written per line when writing to csv.
    #  In case it's correct (on UNIX) to have no '\r' at all, this test
    #  considers either '\r\n' or '\n' as appropriate line endings,
    #  but not '\r\r\n'.
    with tmpdir() as dn:
        ddf.to_csv(os.path.join(dn, 'foo*.csv'), header=False, index=False)
        filename = os.path.join(dn, 'foo0.csv')
        with open(filename, 'rb') as f:
            raw = f.read()
    assert raw in expected
```

Problem statement:
dask DataFrame.to_csv writes extra carriage return on Windows
When using `dask.dataframe.DataFrame.to_csv`, I noticed that lines end with `\r\r\n`, as opposed to simply `\r\n`. I'm guessing this is a Windows issue, although I haven't had the opportunity to test on another system.

The following example shows the bytes produced when using pandas and dask to write the same data to csv. The pandas output is what I was expecting for both. Hopefully this makes it clear what's going on.

Code
-----
```python
import sys
import dask, dask.dataframe as dd
import pandas as pd

df = pd.DataFrame({'a': [0]})
ddf = dd.from_pandas(df, npartitions=1)

df.to_csv('temp-pandas.csv')
pandas_bytes = open('temp-pandas.csv', 'rb').read()

ddf.to_csv('temp-dask_*.csv')
dask_bytes = open('temp-dask_0.csv', 'rb').read()

print('pandas result:', pandas_bytes)
print('dask result  :', dask_bytes)
print('---')
print('dask version:', dask.__version__)
print('Python version:', sys.version)
```

Output
-------
```
pandas result: b',a\r\n0,0\r\n'
dask result  : b',a\r\r\n0,0\r\r\n'
---
dask version: 1.2.2
Python version: 3.7.3 (default, Mar 27 2019, 17:13:21) [MSC v.1915 64 bit (AMD64)]
```

Thanks for the work on an awesome package!

Hints from issue discussion:
What version of pandas do you have? This changed recently, and there were some (unavoidable) issues.
I'm not sure, but https://github.com/pandas-dev/pandas/issues/25048 may be relevant.
My pandas version is 0.24.0
I don't see a direct connection between this and https://github.com/pandas-dev/pandas/issues/25048, as it seems like the issue there is with line endings embedded in the data, not the line endings generated for every record. Additionally, the issue I'm seeing affects dask, but *not* pandas. I could definitely be missing something, though!

I also poked around a bit more, and I came up with a fix to [`dask.bytes.core.OpenFile`](https://github.com/dask/dask/blob/master/dask/bytes/core.py#L192-L194) that seems to work in my case, but may have unintended consequences, for all I know, especially if `OpenFile` uses text mode for file types besides csv. Specifically, `OpenFile.__enter__` needs to pass `newline=''` when it calls `io.TextIOWrapper`. In other words, my issue is solved by changing
```python
        if 't' in self.mode:
            f = io.TextIOWrapper(f, encoding=self.encoding,
                                 errors=self.errors)
```
to 
```python
        if 't' in self.mode:
            f = io.TextIOWrapper(f, encoding=self.encoding,
                                 errors=self.errors, newline='')
```
in [`dask/bytes/core.py`](https://github.com/dask/dask/blob/master/dask/bytes/core.py#L192-L194)

Is it helpful if I submit a PR? Related to that, are there any obvious reasons not to make this change?
Admittedly, I don't really know what I'm doing when it comes to PRs, so it would probably be next week.
    errors=self.errors, newline='')

Obviously, this should be carefully tested on all platforms.

From the TextIOWrapper docs:

> On output, if newline is None, any '\n' characters written are
  translated to the system default line separator, os.linesep. If
  newline is '' or '\n', no translation takes place. If newline is any
  of the other legal values, any '\n' characters written are translated
  to the given string.

Could it be that, on Windows, pandas isn't realising that it should be writing text here, and is pre-converting line-endings?
Note that `pd.DataFrame.to_csv` also has various kwargs to deal with encoding, including `line_terminator=` ("versionchanged: 0.24.0", says the doc); I don't know when this is applied, or how pandas decides if it's writing in text mode or not (perhaps @TomAugspurger knows).
I haven't followed either discussion too closely, so the following may be incorrect.

> I don't see a direct connection between this and pandas-dev/pandas#25048, as it seems like the issue there is with line endings embedded in the data, not the line endings generated for every record. 

> Additionally, the issue I'm seeing affects dask, but not pandas. 

I think that https://github.com/pandas-dev/pandas/pull/25624/files answers those, maybe. I suspect that dask is creating a file object internally, which may need to specify `newline=''`.

I think 
That's really annoying, why change this! I wouldn't want to alter OpenFile directly, other windows dask users, just writing to text files, would have their experience changed.
I think https://github.com/pandas-dev/pandas/issues/25048 has most of the
rationale, but I didn't follow it too closely.

On Fri, May 17, 2019 at 9:22 AM Martin Durant <notifications@github.com>
wrote:

> That's really annoying, why change this! I wouldn't want to alter OpenFile
> directly, other windows dask users, just writing to text files, would have
> their experience changed.
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/dask/dask/issues/4809?email_source=notifications&email_token=AAKAOIUU5FWBMB5K67QM7OLPV25RJA5CNFSM4HNPV47KYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGODVU4QPA#issuecomment-493471804>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/AAKAOIR6PQWBGH7STZCYKETPV25RJANCNFSM4HNPV47A>
> .
>

I am not in agreement with the discussion there, but I don't suppose there's anything to be done about it now. (you *can* tell is a file object is text-mode)

Perhaps, from Dask's point of view, we should always pass `lineterminator='\n'`, in the pandas call, since we know it will passed to TextIOWrapper. Is it right, though, to use windows line endings ever? What if the write is going to somewhere remote?

Commit message: modfied open_files/OpenFile to accept a newline parameter, similar to io.TextIOWrapper or the builtin open on py3. Pass newline='' to open_files when preparing to write csv files.

Fixed #4809

Patch to review:
diff --git a/dask/bytes/core.py b/dask/bytes/core.py
index 7cc4840fa96..833fb544684 100644
--- a/dask/bytes/core.py
+++ b/dask/bytes/core.py
@@ -162,15 +162,18 @@ class OpenFile(object):
         The encoding to use if opened in text mode.
     errors : str or None, optional
         How to handle encoding errors if opened in text mode.
+    newline : None, '', '\n', '\r', or '\r\n'.
+        Passed to TextIOWrapper in text mode
     """
     def __init__(self, fs, path, mode='rb', compression=None, encoding=None,
-                 errors=None):
+                 errors=None, newline=None):
         self.fs = fs
         self.path = path
         self.mode = mode
         self.compression = get_compression(path, compression)
         self.encoding = encoding
         self.errors = errors
+        self.newline = newline
         self.fobjects = []
 
     def __reduce__(self):
@@ -191,7 +194,7 @@ def __enter__(self):
 
         if 't' in self.mode:
             f = io.TextIOWrapper(f, encoding=self.encoding,
-                                 errors=self.errors)
+                                 errors=self.errors, newline=self.newline)
             fobjects.append(f)
 
         self.fobjects = fobjects
@@ -208,7 +211,8 @@ def close(self):
 
 
 def open_files(urlpath, mode='rb', compression=None, encoding='utf8',
-               errors=None, name_function=None, num=1, **kwargs):
+               errors=None, name_function=None, num=1, newline=None,
+               **kwargs):
     """ Given a path or paths, return a list of ``OpenFile`` objects.
 
     Parameters
@@ -232,6 +236,8 @@ def open_files(urlpath, mode='rb', compression=None, encoding='utf8',
     num : int [1]
         if writing mode, number of files we expect to create (passed to
         name+function)
+    newline : None, '', '\n', '\r', or '\r\n'.
+        Passed to TextIOWrapper in text mode
     **kwargs : dict
         Extra options that make sense to a particular storage connection, e.g.
         host, port, username, password, etc.
@@ -250,7 +256,7 @@ def open_files(urlpath, mode='rb', compression=None, encoding='utf8',
                                              storage_options=kwargs)
 
     return [OpenFile(fs, path, mode=mode, compression=compression,
-                     encoding=encoding, errors=errors)
+                     encoding=encoding, errors=errors, newline=newline)
             for path in paths]
 
 
diff --git a/dask/dataframe/io/csv.py b/dask/dataframe/io/csv.py
index d87fcfdcf2a..977b6c688ed 100644
--- a/dask/dataframe/io/csv.py
+++ b/dask/dataframe/io/csv.py
@@ -628,7 +628,8 @@ def to_csv(df, filename, name_function=None, compression=None, compute=True,
 
     files = open_files(filename, compression=compression, mode=mode,
                        encoding=encoding, name_function=name_function,
-                       num=df.npartitions, **(storage_options or {}))
+                       num=df.npartitions, newline='',
+                       **(storage_options or {}))
 
     to_csv_chunk = delayed(_write_csv, pure=False)
 
diff --git a/dask/dataframe/io/tests/test_csv.py b/dask/dataframe/io/tests/test_csv.py
index 33e3c0f3686..1edcbacddfc 100644
--- a/dask/dataframe/io/tests/test_csv.py
+++ b/dask/dataframe/io/tests/test_csv.py
@@ -1275,3 +1275,21 @@ def test_to_csv_header(header, header_first_partition_only, expected_first, expe
             line = fp.readline()
             assert line == expected_next
         os.remove(filename)
+
+def test_to_csv_line_ending():
+    df = pd.DataFrame({'x': [0]})
+    ddf = dd.from_pandas(df, npartitions=1)
+    expected = {b'0\r\n', b'0\n'} # either/or
+    # For comparison...
+    # unexpected = {b'0\r\r\n'}
+    # This test addresses GH4809, and checks that only (at most) one
+    #  '\r' character is written per line when writing to csv.
+    #  In case it's correct (on UNIX) to have no '\r' at all, this test
+    #  considers either '\r\n' or '\n' as appropriate line endings,
+    #  but not '\r\r\n'.
+    with tmpdir() as dn:
+        ddf.to_csv(os.path.join(dn, 'foo*.csv'), header=False, index=False)
+        filename = os.path.join(dn, 'foo0.csv')
+        with open(filename, 'rb') as f:
+            raw = f.read()
+    assert raw in expected


Code review comments:
diff_hunk: @@ -162,15 +162,18 @@ class OpenFile(object):
         The encoding to use if opened in text mode.
     errors : str or None, optional
         How to handle encoding errors if opened in text mode.
+    newline : None, '', '\n', '\r', or '\r\n'.
path: dask/bytes/core.py
text: @user1:
For a fixed set of options, the doc formatting is typically `{None, '', ...}`. And the first one is the default.

@author:
This is updated now.