Fix issue with Hive partitions compatibility. #397

Author: igorborgest

awswrangler/catalog/_add.py

@@ -43,6 +43,7 @@ def add_csv_partitions(
     compression: Optional[str] = None,
     sep: str = ",",
     boto3_session: Optional[boto3.Session] = None,
+    columns_types: Optional[Dict[str, str]] = None,
 ) -> None:
     """Add partitions (metadata) to a CSV Table in the AWS Glue Catalog.
 
@@ -64,6 +65,10 @@ def add_csv_partitions(
         String of length 1. Field delimiter for the output file.
     boto3_session : boto3.Session(), optional
         Boto3 Session. The default boto3 session will be used if boto3_session receive None.
+    columns_types: Optional[Dict[str, str]]
+        Only required for Hive compability.
+        Dictionary with keys as column names and values as data types (e.g. {'col0': 'bigint', 'col1': 'double'}).
+        P.S. Only materialized columns please, not partition columns.
 
     Returns
     -------
@@ -85,7 +90,7 @@ def add_csv_partitions(
 
     """
     inputs: List[Dict[str, Any]] = [
-        _csv_partition_definition(location=k, values=v, compression=compression, sep=sep)
+        _csv_partition_definition(location=k, values=v, compression=compression, sep=sep, columns_types=columns_types)
         for k, v in partitions_values.items()
     ]
     _add_partitions(database=database, table=table, boto3_session=boto3_session, inputs=inputs, catalog_id=catalog_id)
@@ -99,6 +104,7 @@ def add_parquet_partitions(
     catalog_id: Optional[str] = None,
     compression: Optional[str] = None,
     boto3_session: Optional[boto3.Session] = None,
+    columns_types: Optional[Dict[str, str]] = None,
 ) -> None:
     """Add partitions (metadata) to a Parquet Table in the AWS Glue Catalog.
 
@@ -118,6 +124,10 @@ def add_parquet_partitions(
         Compression style (``None``, ``snappy``, ``gzip``, etc).
     boto3_session : boto3.Session(), optional
         Boto3 Session. The default boto3 session will be used if boto3_session receive None.
+    columns_types: Optional[Dict[str, str]]
+        Only required for Hive compability.
+        Dictionary with keys as column names and values as data types (e.g. {'col0': 'bigint', 'col1': 'double'}).
+        P.S. Only materialized columns please, not partition columns.
 
     Returns
     -------
@@ -141,7 +151,7 @@ def add_parquet_partitions(
     table = sanitize_table_name(table=table)
     if partitions_values:
         inputs: List[Dict[str, Any]] = [
-            _parquet_partition_definition(location=k, values=v, compression=compression)
+            _parquet_partition_definition(location=k, values=v, compression=compression, columns_types=columns_types)
             for k, v in partitions_values.items()
         ]
         _add_partitions(

awswrangler/catalog/_create.py

@@ -544,7 +544,7 @@ def create_parquet_table(
     path : str
         Amazon S3 path (e.g. s3://bucket/prefix/).
     columns_types: Dict[str, str]
-        Dictionary with keys as column names and vales as data types (e.g. {'col0': 'bigint', 'col1': 'double'}).
+        Dictionary with keys as column names and values as data types (e.g. {'col0': 'bigint', 'col1': 'double'}).
     partitions_types: Dict[str, str], optional
         Dictionary with keys as partition names and values as data types (e.g. {'col2': 'date'}).
     catalog_id : str, optional
@@ -674,7 +674,7 @@ def create_csv_table(
     path : str
         Amazon S3 path (e.g. s3://bucket/prefix/).
     columns_types: Dict[str, str]
-        Dictionary with keys as column names and vales as data types (e.g. {'col0': 'bigint', 'col1': 'double'}).
+        Dictionary with keys as column names and values as data types (e.g. {'col0': 'bigint', 'col1': 'double'}).
     partitions_types: Dict[str, str], optional
         Dictionary with keys as partition names and values as data types (e.g. {'col2': 'date'}).
     compression : str, optional

awswrangler/catalog/_definitions.py

@@ -38,9 +38,11 @@ def _parquet_table_definition(
     }
 
 
-def _parquet_partition_definition(location: str, values: List[str], compression: Optional[str]) -> Dict[str, Any]:
+def _parquet_partition_definition(
+    location: str, values: List[str], compression: Optional[str], columns_types: Optional[Dict[str, str]]
+) -> Dict[str, Any]:
     compressed: bool = compression is not None
-    return {
+    definition: Dict[str, Any] = {
         "StorageDescriptor": {
             "InputFormat": "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat",
             "OutputFormat": "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat",
@@ -51,9 +53,15 @@ def _parquet_partition_definition(location: str, values: List[str], compression:
                 "SerializationLibrary": "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe",
             },
             "StoredAsSubDirectories": False,
+            "NumberOfBuckets": -1,
         },
         "Values": values,
     }
+    if columns_types is not None:
+        definition["StorageDescriptor"]["Columns"] = [
+            {"Name": cname, "Type": dtype} for cname, dtype in columns_types.items()
+        ]
+    return definition
 
 
 def _csv_table_definition(
@@ -106,9 +114,11 @@ def _csv_table_definition(
     }
 
 
-def _csv_partition_definition(location: str, values: List[str], compression: Optional[str], sep: str) -> Dict[str, Any]:
+def _csv_partition_definition(
+    location: str, values: List[str], compression: Optional[str], sep: str, columns_types: Optional[Dict[str, str]]
+) -> Dict[str, Any]:
     compressed: bool = compression is not None
-    return {
+    definition: Dict[str, Any] = {
         "StorageDescriptor": {
             "InputFormat": "org.apache.hadoop.mapred.TextInputFormat",
             "OutputFormat": "org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat",
@@ -119,6 +129,12 @@ def _csv_partition_definition(location: str, values: List[str], compression: Opt
                 "SerializationLibrary": "org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe",
             },
             "StoredAsSubDirectories": False,
+            "NumberOfBuckets": -1,
         },
         "Values": values,
     }
+    if columns_types is not None:
+        definition["StorageDescriptor"]["Columns"] = [
+            {"Name": cname, "Type": dtype} for cname, dtype in columns_types.items()
+        ]
+    return definition

awswrangler/catalog/_utils.py

@@ -235,7 +235,7 @@ def extract_athena_types(
     Returns
     -------
     Tuple[Dict[str, str], Dict[str, str]]
-        columns_types: Dictionary with keys as column names and vales as
+        columns_types: Dictionary with keys as column names and values as
         data types (e.g. {'col0': 'bigint', 'col1': 'double'}). /
         partitions_types: Dictionary with keys as partition names
         and values as data types (e.g. {'col2': 'date'}).

awswrangler/s3/_read_parquet.py

@@ -762,7 +762,7 @@ def read_parquet_metadata(
     Returns
     -------
     Tuple[Dict[str, str], Optional[Dict[str, str]]]
-        columns_types: Dictionary with keys as column names and vales as
+        columns_types: Dictionary with keys as column names and values as
         data types (e.g. {'col0': 'bigint', 'col1': 'double'}). /
         partitions_types: Dictionary with keys as partition names
         and values as data types (e.g. {'col2': 'date'}).

awswrangler/s3/_write_parquet.py

@@ -562,6 +562,7 @@ def to_parquet(  # pylint: disable=too-many-arguments,too-many-locals
                     compression=compression,
                     boto3_session=session,
                     catalog_id=catalog_id,
+                    columns_types=columns_types,
                 )
     return {"paths": paths, "partitions_values": partitions_values}
 
@@ -700,7 +701,7 @@ def store_parquet_metadata(  # pylint: disable=too-many-arguments
     -------
     Tuple[Dict[str, str], Optional[Dict[str, str]], Optional[Dict[str, List[str]]]]
         The metadata used to create the Glue Table.
-        columns_types: Dictionary with keys as column names and vales as
+        columns_types: Dictionary with keys as column names and values as
         data types (e.g. {'col0': 'bigint', 'col1': 'double'}). /
         partitions_types: Dictionary with keys as partition names
         and values as data types (e.g. {'col2': 'date'}). /
@@ -766,5 +767,6 @@ def store_parquet_metadata(  # pylint: disable=too-many-arguments
             compression=compression,
             boto3_session=session,
             catalog_id=catalog_id,
+            columns_types=columns_types,
         )
     return columns_types, partitions_types, partitions_values

awswrangler/s3/_write_text.py

@@ -442,6 +442,7 @@ def to_csv(  # pylint: disable=too-many-arguments,too-many-locals
                     boto3_session=session,
                     sep=sep,
                     catalog_id=catalog_id,
+                    columns_types=columns_types,
                 )
     return {"paths": paths, "partitions_values": partitions_values}