Repository: pandas-dev/pandas
Pull request: BUG: read_parquet, to_parquet for s3 destinations
- [x] closes #19134
- [x] tests added / passed
- [x] passes `git diff upstream/master -u -- "*.py" | flake8 --diff`

closes #19134 

Problem statement:
to_parquet fails when S3 is the destination
```python
import pandas as pd

df = pd.DataFrame({'field': [1,2,3]})
df.to_parquet("s3://pandas-test/test.parquet", engine='pyarrow')
```
#### Problem description

pandas uses `S3FS` for writing files to S3. `S3File` objects are being opened in `rb` mode.

There are several possible fail cases in the form of an exceptions in the fail chain.

3 different components

 - S3Filesystem, 
 - pyarrow writer, 
 - fastparquet reader & writer.

### pyarrow - write attempt
FileNotFoundError or ValueError (depends on if file exists in S3 or not).

### fastparquet - read attempt
Exception in attempting to concat str and S3File

### fastparquet - write attempt
Exception in attempting to open path using default_open



The above code produces 

```
C:\Users\maxim.veksler\source\DTank\venv\Scripts\python.exe C:/Users/maxim.veksler/source/DTank/DTank/par.py
Traceback (most recent call last):
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 396, in info
    kwargs, Bucket=bucket, Key=key, **self.req_kw)
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 170, in _call_s3
    return method(**additional_kwargs)
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\botocore\client.py", line 314, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\botocore\client.py", line 612, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (404) when calling the HeadObject operation: Not Found

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "c:\users\maxim.veksler\source\pandas\pandas\io\s3.py", line 25, in get_filepath_or_buffer
    filepath_or_buffer = fs.open(_strip_schema(filepath_or_buffer))
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 293, in open
    fill_cache=fill_cache, s3_additional_kwargs=kwargs)
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 931, in __init__
    self.size = self.info()['Size']
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 940, in info
    return self.s3.info(self.path, **kwargs)
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 402, in info
    raise FileNotFoundError(path)
FileNotFoundError: pandas-test/test.parquet

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 396, in info
    kwargs, Bucket=bucket, Key=key, **self.req_kw)
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 170, in _call_s3
    return method(**additional_kwargs)
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\botocore\client.py", line 314, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\botocore\client.py", line 612, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (403) when calling the HeadObject operation: Forbidden

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "C:/Users/maxim.veksler/source/DTank/DTank/par.py", line 4, in <module>
    df.to_parquet("s3://pandas-test/test.parquet", engine='pyarrow')
  File "c:\users\maxim.veksler\source\pandas\pandas\core\frame.py", line 1649, in to_parquet
    compression=compression, **kwargs)
  File "c:\users\maxim.veksler\source\pandas\pandas\io\parquet.py", line 227, in to_parquet
    return impl.write(df, path, compression=compression, **kwargs)
  File "c:\users\maxim.veksler\source\pandas\pandas\io\parquet.py", line 110, in write
    path, _, _ = get_filepath_or_buffer(path)
  File "c:\users\maxim.veksler\source\pandas\pandas\io\common.py", line 202, in get_filepath_or_buffer
    compression=compression)
  File "c:\users\maxim.veksler\source\pandas\pandas\io\s3.py", line 34, in get_filepath_or_buffer
    filepath_or_buffer = fs.open(_strip_schema(filepath_or_buffer))
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 293, in open
    fill_cache=fill_cache, s3_additional_kwargs=kwargs)
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 931, in __init__
    self.size = self.info()['Size']
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 940, in info
    return self.s3.info(self.path, **kwargs)
  File "C:\Users\maxim.veksler\source\DTank\venv\lib\site-packages\s3fs\core.py", line 402, in info
    raise FileNotFoundError(path)
FileNotFoundError: pandas-test/test.parquet

Process finished with exit code 1

```

#### Output of ``pd.show_versions()``

<details>

INSTALLED VERSIONS
------------------
commit: None
python: 3.6.3.final.0
python-bits: 64
OS: Windows
OS-release: 10
machine: AMD64
processor: Intel64 Family 6 Model 142 Stepping 9, GenuineIntel
byteorder: little
LC_ALL: None
LANG: None
LOCALE: None.None

pandas: 0.22.0
pytest: 2.8.0
pip: 9.0.1
setuptools: 38.4.0
Cython: 0.27.3
numpy: 1.13.3
scipy: None
pyarrow: 0.7.1
xarray: None
IPython: None
sphinx: None
patsy: None
dateutil: 2.6.1
pytz: 2017.3
blosc: None
bottleneck: None
tables: None
numexpr: None
feather: None
matplotlib: 2.1.0
openpyxl: None
xlrd: None
xlwt: None
xlsxwriter: None
lxml: None
bs4: None
html5lib: None
sqlalchemy: 1.1.15
pymysql: None
psycopg2: None
jinja2: 2.10
s3fs: 0.1.2
fastparquet: None
pandas_gbq: None
pandas_datareader: None
</details>

Hints from issue discussion:
cc @martindurant
have you tried this the default parquet write (install ``pyarrow``)?
  
You mean bucket does not exist or isn't writable?
In S3, it is not a simple matter to ascertain whether a user has write access to some location, the easiest thing to do often is just to try.
@jreback, @martindurant yes, pyarrow is installed and found.

The issue is that the S3File is being opened in 'rb' mode, please see PR for the details of the fix.

Commit message: Explain why fastparquet is unimplemented.

Patch to review:
diff --git a/doc/source/whatsnew/v0.23.0.txt b/doc/source/whatsnew/v0.23.0.txt
index dc305f36f32ec..2f39c23126f3a 100644
--- a/doc/source/whatsnew/v0.23.0.txt
+++ b/doc/source/whatsnew/v0.23.0.txt
@@ -415,6 +415,7 @@ I/O
 - Bug in :func:`read_sas` where a file with 0 variables gave an ``AttributeError`` incorrectly. Now it gives an ``EmptyDataError`` (:issue:`18184`)
 - Bug in :func:`DataFrame.to_latex()` where pairs of braces meant to serve as invisible placeholders were escaped (:issue:`18667`)
 - Bug in :func:`read_json` where large numeric values were causing an ``OverflowError`` (:issue:`18842`)
+- Bug in :func:`DataFrame.to_parquet` exception is thrown if write destination is S3 (:issue:`19134`)
 -
 
 Plotting
diff --git a/pandas/io/common.py b/pandas/io/common.py
index da60698fe529f..a058eecb431e0 100644
--- a/pandas/io/common.py
+++ b/pandas/io/common.py
@@ -91,14 +91,6 @@ def _is_url(url):
         return False
 
 
-def _is_s3_url(url):
-    """Check for an s3, s3n, or s3a url"""
-    try:
-        return parse_url(url).scheme in ['s3', 's3n', 's3a']
-    except:
-        return False
-
-
 def _expand_user(filepath_or_buffer):
     """Return the argument with an initial component of ~ or ~user
        replaced by that user's home directory.
@@ -168,8 +160,16 @@ def _stringify_path(filepath_or_buffer):
     return filepath_or_buffer
 
 
+def is_s3_url(url):
+    """Check for an s3, s3n, or s3a url"""
+    try:
+        return parse_url(url).scheme in ['s3', 's3n', 's3a']
+    except:
+        return False
+
+
 def get_filepath_or_buffer(filepath_or_buffer, encoding=None,
-                           compression=None):
+                           compression=None, mode=None):
     """
     If the filepath_or_buffer is a url, translate and return the buffer.
     Otherwise passthrough.
@@ -179,6 +179,8 @@ def get_filepath_or_buffer(filepath_or_buffer, encoding=None,
     filepath_or_buffer : a url, filepath (str, py.path.local or pathlib.Path),
                          or buffer
     encoding : the encoding to use to decode py3 bytes, default is 'utf-8'
+    mode : {'rb', 'wb', 'ab'} applies to S3 where a write mandates opening the
+            file in 'wb' mode.
 
     Returns
     -------
@@ -195,11 +197,12 @@ def get_filepath_or_buffer(filepath_or_buffer, encoding=None,
         reader = BytesIO(req.read())
         return reader, encoding, compression
 
-    if _is_s3_url(filepath_or_buffer):
+    if is_s3_url(filepath_or_buffer):
         from pandas.io import s3
         return s3.get_filepath_or_buffer(filepath_or_buffer,
                                          encoding=encoding,
-                                         compression=compression)
+                                         compression=compression,
+                                         mode=mode)
 
     if isinstance(filepath_or_buffer, (compat.string_types,
                                        compat.binary_type,
diff --git a/pandas/io/parquet.py b/pandas/io/parquet.py
index 0c88706a3bec2..3f3f472317637 100644
--- a/pandas/io/parquet.py
+++ b/pandas/io/parquet.py
@@ -5,7 +5,7 @@
 from pandas import DataFrame, RangeIndex, Int64Index, get_option
 from pandas.compat import string_types
 from pandas.core.common import AbstractMethodError
-from pandas.io.common import get_filepath_or_buffer
+from pandas.io.common import get_filepath_or_buffer, is_s3_url
 
 
 def get_engine(engine):
@@ -107,7 +107,7 @@ def write(self, df, path, compression='snappy',
         self.validate_dataframe(df)
         if self._pyarrow_lt_070:
             self._validate_write_lt_070(df)
-        path, _, _ = get_filepath_or_buffer(path)
+        path, _, _ = get_filepath_or_buffer(path, mode='wb')
 
         if self._pyarrow_lt_060:
             table = self.api.Table.from_pandas(df, timestamps_to_ms=True)
@@ -190,6 +190,10 @@ def __init__(self):
         self.api = fastparquet
 
     def write(self, df, path, compression='snappy', **kwargs):
+        if is_s3_url(path):
+            raise NotImplementedError("fastparquet s3 write isn't implemented."
+                                      " Consider using pyarrow instead.")
+
         self.validate_dataframe(df)
         # thriftpy/protocol/compact.py:339:
         # DeprecationWarning: tostring() is deprecated.
diff --git a/pandas/io/s3.py b/pandas/io/s3.py
index 5e48de757d00e..e2650e29c0db3 100644
--- a/pandas/io/s3.py
+++ b/pandas/io/s3.py
@@ -19,10 +19,14 @@ def _strip_schema(url):
 
 
 def get_filepath_or_buffer(filepath_or_buffer, encoding=None,
-                           compression=None):
+                           compression=None, mode=None):
+
+    if mode is None:
+        mode = 'rb'
+
     fs = s3fs.S3FileSystem(anon=False)
     try:
-        filepath_or_buffer = fs.open(_strip_schema(filepath_or_buffer))
+        filepath_or_buffer = fs.open(_strip_schema(filepath_or_buffer), mode)
     except (OSError, NoCredentialsError):
         # boto3 has troubles when trying to access a public file
         # when credentialed...
@@ -31,5 +35,5 @@ def get_filepath_or_buffer(filepath_or_buffer, encoding=None,
         # A NoCredentialsError is raised if you don't have creds
         # for that bucket.
         fs = s3fs.S3FileSystem(anon=True)
-        filepath_or_buffer = fs.open(_strip_schema(filepath_or_buffer))
+        filepath_or_buffer = fs.open(_strip_schema(filepath_or_buffer), mode)
     return filepath_or_buffer, None, compression
diff --git a/pandas/tests/io/test_parquet.py b/pandas/tests/io/test_parquet.py
index 31c2ded49b7a0..287e266cd0084 100644
--- a/pandas/tests/io/test_parquet.py
+++ b/pandas/tests/io/test_parquet.py
@@ -212,28 +212,42 @@ def check_error_on_write(self, df, engine, exc):
             with tm.ensure_clean() as path:
                 to_parquet(df, path, engine, compression=None)
 
-    def check_round_trip(self, df, engine, expected=None,
+    def check_round_trip(self, df, engine, expected=None, path=None,
                          write_kwargs=None, read_kwargs=None,
                          check_names=True):
+
         if write_kwargs is None:
-            write_kwargs = {}
+            write_kwargs = {'compression': None}
+
         if read_kwargs is None:
             read_kwargs = {}
-        with tm.ensure_clean() as path:
-            df.to_parquet(path, engine, **write_kwargs)
-            result = read_parquet(path, engine, **read_kwargs)
 
-            if expected is None:
-                expected = df
-            tm.assert_frame_equal(result, expected, check_names=check_names)
+        if expected is None:
+            expected = df
 
-            # repeat
-            to_parquet(df, path, engine, **write_kwargs)
-            result = pd.read_parquet(path, engine, **read_kwargs)
+        if path is None:
+            with tm.ensure_clean() as path:
+                df.to_parquet(path, engine, **write_kwargs)
+                actual = read_parquet(path, engine, **read_kwargs)
+                tm.assert_frame_equal(expected, actual,
+                                      check_names=check_names)
+
+                # repeat
+                df.to_parquet(path, engine, **write_kwargs)
+                actual = read_parquet(path, engine, **read_kwargs)
+                tm.assert_frame_equal(expected, actual,
+                                      check_names=check_names)
+        else:
+            df.to_parquet(path, engine, **write_kwargs)
+            actual = read_parquet(path, engine, **read_kwargs)
+            tm.assert_frame_equal(expected, actual,
+                                  check_names=check_names)
 
-            if expected is None:
-                expected = df
-            tm.assert_frame_equal(result, expected, check_names=check_names)
+            # repeat
+            df.to_parquet(path, engine, **write_kwargs)
+            actual = read_parquet(path, engine, **read_kwargs)
+            tm.assert_frame_equal(expected, actual,
+                                  check_names=check_names)
 
 
 class TestBasic(Base):
@@ -251,7 +265,7 @@ def test_columns_dtypes(self, engine):
 
         # unicode
         df.columns = [u'foo', u'bar']
-        self.check_round_trip(df, engine, write_kwargs={'compression': None})
+        self.check_round_trip(df, engine)
 
     def test_columns_dtypes_invalid(self, engine):
 
@@ -292,7 +306,6 @@ def test_read_columns(self, engine):
 
         expected = pd.DataFrame({'string': list('abc')})
         self.check_round_trip(df, engine, expected=expected,
-                              write_kwargs={'compression': None},
                               read_kwargs={'columns': ['string']})
 
     def test_write_index(self, engine):
@@ -304,7 +317,7 @@ def test_write_index(self, engine):
                 pytest.skip("pyarrow is < 0.7.0")
 
         df = pd.DataFrame({'A': [1, 2, 3]})
-        self.check_round_trip(df, engine, write_kwargs={'compression': None})
+        self.check_round_trip(df, engine)
 
         indexes = [
             [2, 3, 4],
@@ -315,15 +328,12 @@ def test_write_index(self, engine):
         # non-default index
         for index in indexes:
             df.index = index
-            self.check_round_trip(
-                df, engine,
-                write_kwargs={'compression': None},
-                check_names=check_names)
+            self.check_round_trip(df, engine, check_names=check_names)
 
         # index with meta-data
         df.index = [0, 1, 2]
         df.index.name = 'foo'
-        self.check_round_trip(df, engine, write_kwargs={'compression': None})
+        self.check_round_trip(df, engine)
 
     def test_write_multiindex(self, pa_ge_070):
         # Not suppoprted in fastparquet as of 0.1.3 or older pyarrow version
@@ -332,7 +342,7 @@ def test_write_multiindex(self, pa_ge_070):
         df = pd.DataFrame({'A': [1, 2, 3]})
         index = pd.MultiIndex.from_tuples([('a', 1), ('a', 2), ('b', 1)])
         df.index = index
-        self.check_round_trip(df, engine, write_kwargs={'compression': None})
+        self.check_round_trip(df, engine)
 
     def test_write_column_multiindex(self, engine):
         # column multi-index
@@ -426,6 +436,11 @@ def test_categorical_unsupported(self, pa_lt_070):
         df = pd.DataFrame({'a': pd.Categorical(list('abc'))})
         self.check_error_on_write(df, pa, NotImplementedError)
 
+    def test_s3_roundtrip(self, df_compat, s3_resource, pa):
+        # GH #19134
+        self.check_round_trip(df_compat, pa,
+                              path='s3://pandas-test/pyarrow.parquet')
+
 
 class TestParquetFastParquet(Base):
 
@@ -436,7 +451,7 @@ def test_basic(self, fp, df_full):
         # additional supported types for fastparquet
         df['timedelta'] = pd.timedelta_range('1 day', periods=3)
 
-        self.check_round_trip(df, fp, write_kwargs={'compression': None})
+        self.check_round_trip(df, fp)
 
     @pytest.mark.skip(reason="not supported")
     def test_duplicate_columns(self, fp):
@@ -449,8 +464,7 @@ def test_duplicate_columns(self, fp):
     def test_bool_with_none(self, fp):
         df = pd.DataFrame({'a': [True, None, False]})
         expected = pd.DataFrame({'a': [1.0, np.nan, 0.0]}, dtype='float16')
-        self.check_round_trip(df, fp, expected=expected,
-                              write_kwargs={'compression': None})
+        self.check_round_trip(df, fp, expected=expected)
 
     def test_unsupported(self, fp):
 
@@ -466,7 +480,7 @@ def test_categorical(self, fp):
         if LooseVersion(fastparquet.__version__) < LooseVersion("0.1.3"):
             pytest.skip("CategoricalDtype not supported for older fp")
         df = pd.DataFrame({'a': pd.Categorical(list('abc'))})
-        self.check_round_trip(df, fp, write_kwargs={'compression': None})
+        self.check_round_trip(df, fp)
 
     def test_datetime_tz(self, fp):
         # doesn't preserve tz
@@ -475,8 +489,7 @@ def test_datetime_tz(self, fp):
 
         # warns on the coercion
         with catch_warnings(record=True):
-            self.check_round_trip(df, fp, df.astype('datetime64[ns]'),
-                                  write_kwargs={'compression': None})
+            self.check_round_trip(df, fp, df.astype('datetime64[ns]'))
 
     def test_filter_row_groups(self, fp):
         d = {'a': list(range(0, 3))}
@@ -486,3 +499,11 @@ def test_filter_row_groups(self, fp):
                           row_group_offsets=1)
             result = read_parquet(path, fp, filters=[('a', '==', 0)])
         assert len(result) == 1
+
+    def test_s3_roundtrip(self, df_compat, s3_resource, fp):
+        # GH #19134
+        with pytest.raises(NotImplementedError):
+            # Known limitation, fastparquet doesn't support writing to S3.
+            # https://git.io/vNCcz
+            self.check_round_trip(df_compat, fp,
+                                  path='s3://pandas-test/fastparquet.parquet')
diff --git a/pandas/tests/io/test_s3.py b/pandas/tests/io/test_s3.py
index 8c2a32af33765..7a3062f470ce8 100644
--- a/pandas/tests/io/test_s3.py
+++ b/pandas/tests/io/test_s3.py
@@ -1,8 +1,8 @@
-from pandas.io.common import _is_s3_url
+from pandas.io.common import is_s3_url
 
 
 class TestS3URL(object):
 
     def test_is_s3_url(self):
-        assert _is_s3_url("s3://pandas/somethingelse.com")
-        assert not _is_s3_url("s4://pandas/somethingelse.com")
+        assert is_s3_url("s3://pandas/somethingelse.com")
+        assert not is_s3_url("s4://pandas/somethingelse.com")


Code review comments:
diff_hunk: @@ -190,6 +190,10 @@ def __init__(self):
         self.api = fastparquet
 
     def write(self, df, path, compression='snappy', **kwargs):
+        if is_s3_url(path):
path: pandas/io/parquet.py
text: @user1:
In this if block, go from `S3File` -> url

```python
path = 's3://{}'.format(path.path)
kwargs['open_with'] = path.s3.open
```

See if that works?