Merge pull request #128 from awslabs/pandas-v1

Bumping Pandas version to 1.0.0

Author: igorborgest

awswrangler/data_types.py

@@ -20,7 +20,7 @@ def athena2pandas(dtype: str) -> str:
     elif dtype == "boolean":
         return "bool"
     elif dtype in ("string", "char", "varchar"):
-        return "str"
+        return "string"
     elif dtype in ("timestamp", "timestamp with time zone"):
         return "datetime64"
     elif dtype == "date":
@@ -117,6 +117,8 @@ def pandas2athena(dtype: str) -> str:
         return "double"
     elif dtype == "bool":
         return "boolean"
+    elif dtype == "string":
+        return "string"
     elif dtype == "object":
         return "string"
     elif dtype.startswith("datetime64"):
@@ -137,7 +139,9 @@ def pandas2redshift(dtype: str, varchar_length: int = 256) -> str:
         return "FLOAT8"
     elif dtype == "bool":
         return "BOOLEAN"
-    elif dtype == "object" and isinstance(dtype, str):
+    elif dtype == "string":
+        return f"VARCHAR({varchar_length})"
+    elif dtype == "object":
         return f"VARCHAR({varchar_length})"
     elif dtype[:10] == "datetime64":
         return "TIMESTAMP"
@@ -375,11 +379,13 @@ def extract_pyarrow_schema_from_pandas(dataframe: pd.DataFrame,
     if indexes_position not in ("right", "left"):
         raise ValueError(f"indexes_position must be \"right\" or \"left\"")
 
-    # Handle exception data types (e.g. Int64)
+    # Handle exception data types (e.g. Int64, string)
     for name, dtype in dataframe.dtypes.to_dict().items():
         dtype = str(dtype)
         if dtype == "Int64":
             cols_dtypes[name] = "int64"
+        elif dtype == "string":
+            cols_dtypes[name] = "string"
         else:
             cols.append(name)
 

awswrangler/glue.py

@@ -235,12 +235,9 @@ def _build_schema(
             preserve_index: bool,
             indexes_position: str,
             cast_columns: Optional[Dict[str, str]] = None) -> Tuple[List[Tuple[str, str]], List[Tuple[str, str]]]:
-        if cast_columns is None:
-            cast_columns = {}
+        cast_columns = {} if cast_columns is None else cast_columns
+        partition_cols = [] if partition_cols is None else partition_cols
         logger.debug(f"dataframe.dtypes:\n{dataframe.dtypes}")
-        if partition_cols is None:
-            partition_cols = []
-
         pyarrow_schema: List[Tuple[str, Any]] = data_types.extract_pyarrow_schema_from_pandas(
             dataframe=dataframe, preserve_index=preserve_index, indexes_position=indexes_position)
 

requirements-dev.txt

@@ -3,9 +3,9 @@ mypy~=0.761
 flake8~=3.7.9
 pytest-cov~=2.8.1
 scikit-learn~=0.22.1
-cfn-lint~=0.27.2
+cfn-lint~=0.27.3
 twine~=3.1.1
-wheel~=0.34.0
+wheel~=0.34.2
 sphinx~=2.3.1
 pyspark~=2.4.4
 pyspark-stubs~=2.4.0.post7

requirements.txt

@@ -1,5 +1,5 @@
 numpy~=1.18.1
-pandas~=0.25.3
+pandas~=1.0.0
 pyarrow~=0.15.1
 botocore>=1.13.34
 boto3>=1.10.34

testing/test_awswrangler/test_pandas.py

@@ -2369,6 +2369,7 @@ def test_s3_overall_nan(bucket, database):
 
 def test_aurora_postgres_load_varchar(bucket, postgres_parameters):
     df = pd.DataFrame({"id": [1, 2, 3], "varchar3": ["foo", "boo", "bar"], "varchar1": ["a", "b", "c"]})
+    df["varchar3"] = df["varchar3"].astype("string")
     path = f"s3://{bucket}/test_aurora_postgres_load_varchar"
     wr.pandas.to_aurora(dataframe=df,
                         connection="aws-data-wrangler-postgres",
@@ -2404,6 +2405,7 @@ def test_aurora_postgres_load_varchar(bucket, postgres_parameters):
 
 def test_aurora_mysql_load_varchar(bucket):
     df = pd.DataFrame({"id": [1, 2, 3], "varchar3": ["foo", "boo", "bar"], "varchar1": ["a", "b", "c"]})
+    df["varchar3"] = df["varchar3"].astype("string")
     path = f"s3://{bucket}/test_aurora_mysql_load_varchar"
     wr.pandas.to_aurora(dataframe=df,
                         connection="aws-data-wrangler-mysql",
@@ -2430,3 +2432,51 @@ def test_aurora_mysql_load_varchar(bucket):
         assert rows[1][2] == "b"
         assert rows[2][2] == "c"
     conn.close()
+
+
+def test_to_parquet_string(bucket, database):
+    path = f"s3://{bucket}/test_to_parquet_string"
+    wr.s3.delete_objects(path=path)
+    df = pd.DataFrame({
+        "id": [1, 2, 3, 4, 5],
+        "c_str": ["foo", None, None, "bar", None],
+    })
+    df["id"] = df["id"].astype("Int64")
+    df["c_str"] = df["c_str"].astype("string")
+    wr.pandas.to_parquet(dataframe=df,
+                         database=database,
+                         path=path,
+                         mode="overwrite",
+                         preserve_index=False,
+                         procs_cpu_bound=5,
+                         inplace=False)
+    sleep(15)
+    df2 = wr.pandas.read_sql_athena(database=database,
+                                    sql="SELECT * FROM test_to_parquet_string ORDER BY id",
+                                    ctas_approach=False)
+    wr.s3.delete_objects(path=path)
+    assert df.equals(df2)
+
+
+def test_to_csv_string(bucket, database):
+    path = f"s3://{bucket}/test_to_csv_string"
+    wr.s3.delete_objects(path=path)
+    df = pd.DataFrame({
+        "id": [1, 2, 3, 4, 5],
+        "c_str": ["foo", None, None, "bar", None],
+    })
+    df["id"] = df["id"].astype("Int64")
+    df["c_str"] = df["c_str"].astype("string")
+    wr.pandas.to_parquet(dataframe=df,
+                         database=database,
+                         path=path,
+                         mode="overwrite",
+                         preserve_index=False,
+                         procs_cpu_bound=5,
+                         inplace=False)
+    sleep(5)
+    df2 = wr.pandas.read_sql_athena(database=database,
+                                    sql="SELECT * FROM test_to_csv_string ORDER BY id",
+                                    ctas_approach=False)
+    wr.s3.delete_objects(path=path)
+    assert df.equals(df2)

testing/test_awswrangler/test_redshift.py

@@ -870,6 +870,7 @@ def test_spectrum_csv(bucket, glue_database, external_schema):
 
 def test_to_redshift_pandas_varchar(bucket, redshift_parameters):
     df = pd.DataFrame({"id": [1, 2, 3], "varchar3": ["foo", "boo", "bar"], "varchar1": ["a", "b", "c"]})
+    df["varchar3"] = df["varchar3"].astype("string")
     path = f"s3://{bucket}/test_to_redshift_pandas_varchar"
     wr.pandas.to_redshift(dataframe=df,
                           path=path,