Bumping version to 0.0.10

Author: igorborgest

.gitignore

@@ -134,6 +134,6 @@ python/
 
 # SAM
 .aws-sam
-testing/*parameters-*.json
+testing/*parameters-*.properties
 testing/*requirements*.txt
 building/*requirements*.txt

README.md

@@ -2,11 +2,11 @@
 
 > Utility belt to handle data on AWS.
 
-[![Release](https://img.shields.io/badge/release-0.0.9-brightgreen.svg)](https://pypi.org/project/awswrangler/)
+[![Release](https://img.shields.io/badge/release-0.0.10-brightgreen.svg)](https://pypi.org/project/awswrangler/)
 [![Downloads](https://img.shields.io/pypi/dm/awswrangler.svg)](https://pypi.org/project/awswrangler/)
 [![Python Version](https://img.shields.io/badge/python-3.6%20%7C%203.7-brightgreen.svg)](https://pypi.org/project/awswrangler/)
 [![Documentation Status](https://readthedocs.org/projects/aws-data-wrangler/badge/?version=latest)](https://aws-data-wrangler.readthedocs.io/en/latest/?badge=latest)
-[![Coverage](https://img.shields.io/badge/coverage-83%25-brightgreen.svg)](https://pypi.org/project/awswrangler/)
+[![Coverage](https://img.shields.io/badge/coverage-87%25-brightgreen.svg)](https://pypi.org/project/awswrangler/)
 [![Average time to resolve an issue](http://isitmaintained.com/badge/resolution/awslabs/aws-data-wrangler.svg)](http://isitmaintained.com/project/awslabs/aws-data-wrangler "Average time to resolve an issue")
 [![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)
 
@@ -35,7 +35,8 @@
 
 ### PySpark
 * PySpark -> Redshift (Parallel)
-* Register Glue table from Dataframe stored on S3 (NEW :star:)
+* Register Glue table from Dataframe stored on S3
+* Flatten nested DataFrames (NEW :star:)
 
 ### General
 * List S3 objects (Parallel)
@@ -45,7 +46,12 @@
 * Copy listed S3 objects (Parallel)
 * Get the size of S3 objects (Parallel)
 * Get CloudWatch Logs Insights query results
-* Load partitions on Athena/Glue table (repair table) (NEW :star:)
+* Load partitions on Athena/Glue table (repair table)
+* Create EMR cluster (For humans) (NEW :star:)
+* Terminate EMR cluster (NEW :star:)
+* Get EMR cluster state (NEW :star:)
+* Submit EMR step (For humans) (NEW :star:)
+* Ger EMR step state (NEW :star:)
 
 ## Installation
 
@@ -195,6 +201,16 @@ session.spark.create_glue_table(dataframe=dataframe,
                                 database="my_database")
 ```
 
+#### Flatten nested PySpark DataFrame
+
+```py3
+session = awswrangler.Session(spark_session=spark)
+dfs = session.spark.flatten(df=df_nested)
+for name, df_flat in dfs:
+    print(name)
+    df_flat.show()
+```
+
 ### General
 
 #### Deleting a bunch of S3 objects (parallel)
@@ -221,6 +237,51 @@ session = awswrangler.Session()
 session.athena.repair_table(database="db_name", table="tbl_name")
 ```
 
+#### Create EMR cluster
+
+```py3
+session = awswrangler.Session()
+cluster_id = session.emr.create_cluster(
+    cluster_name="wrangler_cluster",
+    logging_s3_path=f"s3://BUCKET_NAME/emr-logs/",
+    emr_release="emr-5.27.0",
+    subnet_id="SUBNET_ID",
+    emr_ec2_role="EMR_EC2_DefaultRole",
+    emr_role="EMR_DefaultRole",
+    instance_type_master="m5.xlarge",
+    instance_type_core="m5.xlarge",
+    instance_type_task="m5.xlarge",
+    instance_ebs_size_master=50,
+    instance_ebs_size_core=50,
+    instance_ebs_size_task=50,
+    instance_num_on_demand_master=1,
+    instance_num_on_demand_core=1,
+    instance_num_on_demand_task=1,
+    instance_num_spot_master=0,
+    instance_num_spot_core=1,
+    instance_num_spot_task=1,
+    spot_bid_percentage_of_on_demand_master=100,
+    spot_bid_percentage_of_on_demand_core=100,
+    spot_bid_percentage_of_on_demand_task=100,
+    spot_provisioning_timeout_master=5,
+    spot_provisioning_timeout_core=5,
+    spot_provisioning_timeout_task=5,
+    spot_timeout_to_on_demand_master=True,
+    spot_timeout_to_on_demand_core=True,
+    spot_timeout_to_on_demand_task=True,
+    python3=True,
+    spark_glue_catalog=True,
+    hive_glue_catalog=True,
+    presto_glue_catalog=True,
+    bootstraps_paths=None,
+    debugging=True,
+    applications=["Hadoop", "Spark", "Ganglia", "Hive"],
+    visible_to_all_users=True,
+    key_pair_name=None,
+)
+print(cluster_id)
+```
+
 ## Diving Deep
 
 

awswrangler/__init__.py

@@ -9,6 +9,7 @@
 from awswrangler.cloudwatchlogs import CloudWatchLogs  # noqa
 from awswrangler.glue import Glue  # noqa
 from awswrangler.redshift import Redshift  # noqa
+from awswrangler.emr import EMR  # noqa
 import awswrangler.utils  # noqa
 import awswrangler.data_types  # noqa
 

awswrangler/__version__.py

@@ -1,4 +1,4 @@
 __title__ = "awswrangler"
 __description__ = "Utility belt to handle data on AWS."
-__version__ = "0.0.9"
+__version__ = "0.0.10"
 __license__ = "Apache License 2.0"

awswrangler/athena.py

@@ -15,14 +15,10 @@
 class Athena:
     def __init__(self, session):
         self._session = session
-        self._client_athena = session.boto3_session.client(
-            service_name="athena", config=session.botocore_config
-        )
+        self._client_athena = session.boto3_session.client(service_name="athena", config=session.botocore_config)
 
     def get_query_columns_metadata(self, query_execution_id):
-        response = self._client_athena.get_query_results(
-            QueryExecutionId=query_execution_id, MaxResults=1
-        )
+        response = self._client_athena.get_query_results(QueryExecutionId=query_execution_id, MaxResults=1)
         col_info = response["ResultSet"]["ResultSetMetadata"]["ColumnInfo"]
         return {x["Name"]: x["Type"] for x in col_info}
 
@@ -54,11 +50,8 @@ def create_athena_bucket(self):
 
         :return: Bucket s3 path (E.g. s3://aws-athena-query-results-ACCOUNT-REGION/)
         """
-        account_id = (
-            self._session.boto3_session.client(
-                service_name="sts", config=self._session.botocore_config
-            ).get_caller_identity().get("Account")
-        )
+        account_id = (self._session.boto3_session.client(
+            service_name="sts", config=self._session.botocore_config).get_caller_identity().get("Account"))
         session_region = self._session.boto3_session.region_name
         s3_output = f"s3://aws-athena-query-results-{account_id}-{session_region}/"
         s3_resource = self._session.boto3_session.resource("s3")
@@ -80,12 +73,10 @@ def run_query(self, query, database, s3_output=None, workgroup=None):
         if workgroup is None:
             workgroup = self._session.athena_workgroup
         logger.debug(f"Workgroup: {workgroup}")
-        response = self._client_athena.start_query_execution(
-            QueryString=query,
-            QueryExecutionContext={"Database": database},
-            ResultConfiguration={"OutputLocation": s3_output},
-            WorkGroup=workgroup
-        )
+        response = self._client_athena.start_query_execution(QueryString=query,
+                                                             QueryExecutionContext={"Database": database},
+                                                             ResultConfiguration={"OutputLocation": s3_output},
+                                                             WorkGroup=workgroup)
         return response["QueryExecutionId"]
 
     def wait_query(self, query_execution_id):
@@ -103,9 +94,7 @@ def wait_query(self, query_execution_id):
             response = self._client_athena.get_query_execution(QueryExecutionId=query_execution_id)
             state = response["QueryExecution"]["Status"]["State"]
         logger.debug(f"state: {state}")
-        logger.debug(
-            f"StateChangeReason: {response['QueryExecution']['Status'].get('StateChangeReason')}"
-        )
+        logger.debug(f"StateChangeReason: {response['QueryExecution']['Status'].get('StateChangeReason')}")
         if state == "FAILED":
             raise QueryFailed(response["QueryExecution"]["Status"].get("StateChangeReason"))
         elif state == "CANCELLED":
@@ -129,17 +118,13 @@ def repair_table(self, database, table, s3_output=None, workgroup=None):
         :return: Query execution ID
         """
         query = f"MSCK REPAIR TABLE {table};"
-        query_id = self.run_query(
-            query=query, database=database, s3_output=s3_output, workgroup=workgroup
-        )
+        query_id = self.run_query(query=query, database=database, s3_output=s3_output, workgroup=workgroup)
         self.wait_query(query_execution_id=query_id)
         return query_id
 
     @staticmethod
     def _normalize_name(name):
-        name = "".join(
-            c for c in unicodedata.normalize("NFD", name) if unicodedata.category(c) != "Mn"
-        )
+        name = "".join(c for c in unicodedata.normalize("NFD", name) if unicodedata.category(c) != "Mn")
         name = name.replace(" ", "_")
         name = name.replace("-", "_")
         name = name.replace(".", "_")

awswrangler/cloudwatchlogs.py

@@ -12,18 +12,14 @@
 class CloudWatchLogs:
     def __init__(self, session):
         self._session = session
-        self._client_logs = session.boto3_session.client(
-            service_name="logs", config=session.botocore_config
-        )
+        self._client_logs = session.boto3_session.client(service_name="logs", config=session.botocore_config)
 
-    def start_query(
-        self,
-        query,
-        log_group_names,
-        start_time=datetime(year=1970, month=1, day=1),
-        end_time=datetime.utcnow(),
-        limit=None
-    ):
+    def start_query(self,
+                    query,
+                    log_group_names,
+                    start_time=datetime(year=1970, month=1, day=1),
+                    end_time=datetime.utcnow(),
+                    limit=None):
         """
         Run a query against AWS CloudWatchLogs Insights and wait the results
         https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/CWL_QuerySyntax.html
@@ -72,14 +68,12 @@ def wait_query(self, query_id):
             raise QueryCancelled(f"query ID: {query_id}")
         return response
 
-    def query(
-        self,
-        query,
-        log_group_names,
-        start_time=datetime(year=1970, month=1, day=1),
-        end_time=datetime.utcnow(),
-        limit=None
-    ):
+    def query(self,
+              query,
+              log_group_names,
+              start_time=datetime(year=1970, month=1, day=1),
+              end_time=datetime.utcnow(),
+              limit=None):
         """
         Run a query against AWS CloudWatchLogs Insights and wait the results
         https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/CWL_QuerySyntax.html
@@ -91,12 +85,10 @@ def query(
         :param limit: The maximum number of log events to return in the query.
         :return: Results
         """
-        query_id = self.start_query(
-            query=query,
-            log_group_names=log_group_names,
-            start_time=start_time,
-            end_time=end_time,
-            limit=limit
-        )
+        query_id = self.start_query(query=query,
+                                    log_group_names=log_group_names,
+                                    start_time=start_time,
+                                    end_time=end_time,
+                                    limit=limit)
         response = self.wait_query(query_id=query_id)
         return response["results"]

awswrangler/data_types.py

@@ -294,9 +294,8 @@ def convert_schema(func: Callable, schema: List[Tuple[str, str]]) -> Dict[str, s
     return {name: func(dtype) for name, dtype in schema}
 
 
-def extract_pyarrow_schema_from_pandas(
-    dataframe: pd.DataFrame, preserve_index: bool, indexes_position: str = "right"
-) -> List[Tuple[str, str]]:
+def extract_pyarrow_schema_from_pandas(dataframe: pd.DataFrame, preserve_index: bool,
+                                       indexes_position: str = "right") -> List[Tuple[str, str]]:
     """
     Extract the related Pyarrow schema from any Pandas DataFrame