COMPAT: handle pyarrow deprecation of timestamps_to_ms in .from_pandas with pyarrow < 0.6.0 (#17447)

closes #17438

Author: jreback

Diff for: ci/requirements-3.5.sh

@@ -8,4 +8,4 @@ echo "install 35"
 conda remove -n pandas python-dateutil --force
 pip install python-dateutil
 
-conda install -n pandas -c conda-forge feather-format pyarrow=0.4.1
+conda install -n pandas -c conda-forge feather-format pyarrow=0.5.0

Diff for: doc/source/whatsnew/v0.21.0.txt

@@ -125,7 +125,7 @@ Other Enhancements
 - :func:`DataFrame.select_dtypes` now accepts scalar values for include/exclude as well as list-like. (:issue:`16855`)
 - :func:`date_range` now accepts 'YS' in addition to 'AS' as an alias for start of year (:issue:`9313`)
 - :func:`date_range` now accepts 'Y' in addition to 'A' as an alias for end of year (:issue:`9313`)
-- Integration with `Apache Parquet <https://parquet.apache.org/>`__, including a new top-level :func:`read_parquet` and :func:`DataFrame.to_parquet` method, see :ref:`here <io.parquet>`.
+- Integration with `Apache Parquet <https://parquet.apache.org/>`__, including a new top-level :func:`read_parquet` and :func:`DataFrame.to_parquet` method, see :ref:`here <io.parquet>`. (:issue:`15838`, :issue:`17438`)
 - :func:`DataFrame.add_prefix` and :func:`DataFrame.add_suffix` now accept strings containing the '%' character. (:issue:`17151`)
 - `read_*` methods can now infer compression from non-string paths, such as ``pathlib.Path`` objects (:issue:`17206`).
 - :func:`pd.read_sas()` now recognizes much more of the most frequently used date (datetime) formats in SAS7BDAT files (:issue:`15871`).

Diff for: pandas/io/parquet.py

@@ -58,13 +58,23 @@ def __init__(self):
                               "\nor via pip\n"
                               "pip install -U pyarrow\n")
 
+        self._pyarrow_lt_050 = LooseVersion(pyarrow.__version__) < '0.5.0'
+        self._pyarrow_lt_060 = LooseVersion(pyarrow.__version__) < '0.6.0'
         self.api = pyarrow
 
-    def write(self, df, path, compression='snappy', **kwargs):
+    def write(self, df, path, compression='snappy',
+              coerce_timestamps='ms', **kwargs):
         path, _, _ = get_filepath_or_buffer(path)
-        table = self.api.Table.from_pandas(df, timestamps_to_ms=True)
-        self.api.parquet.write_table(
-            table, path, compression=compression, **kwargs)
+        if self._pyarrow_lt_060:
+            table = self.api.Table.from_pandas(df, timestamps_to_ms=True)
+            self.api.parquet.write_table(
+                table, path, compression=compression, **kwargs)
+
+        else:
+            table = self.api.Table.from_pandas(df)
+            self.api.parquet.write_table(
+                table, path, compression=compression,
+                coerce_timestamps=coerce_timestamps, **kwargs)
 
     def read(self, path):
         path, _, _ = get_filepath_or_buffer(path)