Repository: dask/dask

Issue:
Can't specify storage_options for bag.to_textfiles, dataframe.to_csv
The read functions allow specifying `storage_options` for things like AWS key/secret. The `to_parquet` function allows this as well.

However in other cases such as `bag.to_textfiles()` and `dataframe.to_csv()`, it is not possible to specify these options. It would be nice for there to be consistency, since there is no other way to specify some of the options at this time.

Discussion:
jcrist: > Unsure how to test this

Any way to test using moto, and skip if moto not installed?

If this is hard to test, I'd also be fine merging without.
martindurant: Yeah, moto doesn't disallow writing, even when anonymous, so I don't know what parameters I could pass to s3fs that would have any effect, where as local writes take no parameters at all.

Patch:
diff --git a/dask/array/fft.py b/dask/array/fft.py
index cfccda548..2f64c94ee 100644
--- a/dask/array/fft.py
+++ b/dask/array/fft.py
@@ -228,9 +228,9 @@ def _fftfreq_helper(n, d=1.0, chunks=None, real=False):
     s = n // 2 + 1 if real else n
     t = l - s
 
-    chunks = _normalize_chunks(chunks, (s,))[0] + (t,)
+    chunks = _normalize_chunks(chunks, (s,))
 
-    r = _linspace(0, 1, l, chunks=chunks)
+    r = _linspace(0, 1, l, chunks=(chunks[0] + (t,),))
 
     if real:
         n_2 = n // 2 + 1
@@ -239,6 +239,9 @@ def _fftfreq_helper(n, d=1.0, chunks=None, real=False):
         n_2 = (n + 1) // 2
         r = _concatenate([r[:n_2], r[n_2:-1] - 1])
 
+    if r.chunks != chunks:
+        r = r.rechunk(chunks)
+
     r /= d
 
     return r
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 1cb43c01b..6cc7a8ab6 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -118,7 +118,8 @@ def optimize(dsk, keys, fuse_keys=None, rename_fused_keys=True, **kwargs):
 
 
 def to_textfiles(b, path, name_function=None, compression='infer',
-                 encoding=system_encoding, compute=True, get=None):
+                 encoding=system_encoding, compute=True, get=None,
+                 storage_options=None):
     """ Write bag to disk, one filename per partition, one line per element
 
     **Paths**: This will create one file for each partition in your bag. You
@@ -174,7 +175,7 @@ def to_textfiles(b, path, name_function=None, compression='infer',
     """
     from dask import delayed
     writes = write_bytes(b.to_delayed(), path, name_function, compression,
-                         encoding=encoding)
+                         encoding=encoding, **(storage_options or {}))
 
     # Use Bag optimizations on these delayed objects
     dsk = ensure_dict(delayed(writes).dask)
@@ -642,9 +643,10 @@ class Bag(Base):
 
     @wraps(to_textfiles)
     def to_textfiles(self, path, name_function=None, compression='infer',
-                     encoding=system_encoding, compute=True, get=None):
+                     encoding=system_encoding, compute=True, get=None,
+                     storage_options=None):
         return to_textfiles(self, path, name_function, compression, encoding,
-                            compute, get=get)
+                            compute, get=get, storage_options=storage_options)
 
     def fold(self, binop, combine=None, initial=no_default, split_every=None):
         """ Parallelizable reduction
diff --git a/dask/dataframe/io/csv.py b/dask/dataframe/io/csv.py
index 9ceaad2e2..c999aa414 100644
--- a/dask/dataframe/io/csv.py
+++ b/dask/dataframe/io/csv.py
@@ -364,7 +364,7 @@ def _to_csv_chunk(df, **kwargs):
 
 
 def to_csv(df, filename, name_function=None, compression=None, compute=True,
-           get=None, **kwargs):
+           get=None, storage_options=None, **kwargs):
     """
     Store Dask DataFrame to CSV files
 
@@ -472,10 +472,12 @@ def to_csv(df, filename, name_function=None, compression=None, compute=True,
     decimal: string, default '.'
         Character recognized as decimal separator. E.g. use ',' for
         European data
+    storage_options: dict
+        Parameters passed on to the backend filesystem class.
     """
     values = [_to_csv_chunk(d, **kwargs) for d in df.to_delayed()]
     values = write_bytes(values, filename, name_function, compression,
-                         encoding=None)
+                         encoding=None, **(storage_options or {}))
 
     if compute:
         delayed(values).compute(get=get)


Test patch:
diff --git a/dask/array/tests/test_fft.py b/dask/array/tests/test_fft.py
index c02b7fefd..ab6de9560 100644
--- a/dask/array/tests/test_fft.py
+++ b/dask/array/tests/test_fft.py
@@ -9,6 +9,10 @@ import dask.array.fft
 from dask.array.fft import fft_wrap
 from dask.array.utils import assert_eq
 
+from dask.array.core import (
+    normalize_chunks as _normalize_chunks,
+)
+
 
 def same_keys(a, b):
     def key(k):
@@ -202,7 +206,13 @@ def test_wrap_fftns(modname, funcname, dtype):
 @pytest.mark.parametrize("c", [lambda m: m, lambda m: (1, m - 1)])
 def test_fftfreq(n, d, c):
     c = c(n)
-    assert_eq(da.fft.fftfreq(n, d, chunks=c), np.fft.fftfreq(n, d))
+
+    r1 = np.fft.fftfreq(n, d)
+    r2 = da.fft.fftfreq(n, d, chunks=c)
+
+    assert _normalize_chunks(c, r2.shape) == r2.chunks
+
+    assert_eq(r1, r2)
 
 
 @pytest.mark.parametrize("n", [1, 2, 3, 6, 7])
@@ -210,9 +220,13 @@ def test_fftfreq(n, d, c):
 @pytest.mark.parametrize("c", [lambda m: m // 2 + 1, lambda m: (1, m // 2)])
 def test_rfftfreq(n, d, c):
     c = c(n)
-    assert_eq(
-        da.fft.rfftfreq(n, d, chunks=c), np.fft.rfftfreq(n, d)
-    )
+
+    r1 = np.fft.rfftfreq(n, d)
+    r2 = da.fft.rfftfreq(n, d, chunks=c)
+
+    assert _normalize_chunks(c, r2.shape) == r2.chunks
+
+    assert_eq(r1, r2)
 
 
 @pytest.mark.parametrize("funcname", ["fftshift", "ifftshift"])