Add support for EMR consistent view

Author: igorborgest

awswrangler/__init__.py

@@ -1,6 +1,7 @@
-from typing import Optional
-import logging
 import importlib
+from logging import getLogger, NullHandler
+from typing import Optional
+from sys import version_info
 
 from awswrangler.__version__ import __title__, __description__, __version__  # noqa
 from awswrangler.session import Session  # noqa
@@ -41,7 +42,7 @@ def __getattr__(self, name):
         return getattr(getattr(DynamicInstantiate.__default_session, self._module_name), name)
 
 
-if importlib.util.find_spec("pyspark"):  # type: ignore
+if version_info < (3, 8) and importlib.util.find_spec("pyspark"):  # type: ignore
     from awswrangler.spark import Spark  # noqa
     spark: Spark = DynamicInstantiate("spark", Spark)  # type: ignore
 
@@ -56,4 +57,4 @@ def __getattr__(self, name):
 sagemaker: SageMaker = DynamicInstantiate("sagemaker", SageMaker)  # type: ignore
 cloudwatchlogs: CloudWatchLogs = DynamicInstantiate("cloudwatchlogs", CloudWatchLogs)  # type: ignore
 
-logging.getLogger("awswrangler").addHandler(logging.NullHandler())
+getLogger("awswrangler").addHandler(NullHandler())

awswrangler/emr.py

@@ -122,7 +122,7 @@ def _build_cluster_args(**pars):
                     "Configurations": []
                 }]
             })
-        if pars["spark_glue_catalog"]:
+        if pars["spark_glue_catalog"] is True:
             args["Configurations"].append({
                 "Classification": "spark-hive-site",
                 "Properties": {
@@ -131,7 +131,7 @@ def _build_cluster_args(**pars):
                 },
                 "Configurations": []
             })
-        if pars["hive_glue_catalog"]:
+        if pars["hive_glue_catalog"] is True:
             args["Configurations"].append({
                 "Classification": "hive-site",
                 "Properties": {
@@ -140,15 +140,25 @@ def _build_cluster_args(**pars):
                 },
                 "Configurations": []
             })
-        if pars["presto_glue_catalog"]:
+        if pars["presto_glue_catalog"] is True:
             args["Configurations"].append({
                 "Classification": "presto-connector-hive",
                 "Properties": {
                     "hive.metastore.glue.datacatalog.enabled": "true"
                 },
                 "Configurations": []
             })
-        if pars["maximize_resource_allocation"]:
+        if pars["consistent_view"] is True:
+            args["Configurations"].append({
+                "Classification": "emrfs-site",
+                "Properties": {
+                    "fs.s3.consistent.retryPeriodSeconds": str(pars.get("consistent_view_retry_seconds", "10")),
+                    "fs.s3.consistent": "true",
+                    "fs.s3.consistent.retryCount": str(pars.get("consistent_view_retry_count", "5")),
+                    "fs.s3.consistent.metadata.tableName": pars.get("consistent_view_table_name", "EmrFSMetadata")
+                }
+            })
+        if pars["maximize_resource_allocation"] is True:
             args["Configurations"].append({
                 "Classification": "spark",
                 "Properties": {
@@ -351,6 +361,10 @@ def create_cluster(self,
                        spark_glue_catalog: bool = True,
                        hive_glue_catalog: bool = True,
                        presto_glue_catalog: bool = True,
+                       consistent_view: bool = False,
+                       consistent_view_retry_seconds: int = 10,
+                       consistent_view_retry_count: int = 5,
+                       consistent_view_table_name: str = "EmrFSMetadata",
                        bootstraps_paths: Optional[List[str]] = None,
                        debugging: bool = True,
                        applications: Optional[List[str]] = None,
@@ -369,7 +383,7 @@ def create_cluster(self,
                        steps: Optional[List[Dict[str, Collection[str]]]] = None,
                        keep_cluster_alive_when_no_steps: bool = True,
                        termination_protected: bool = False,
-                       tags: Optional[Dict[str, str]] = None):
+                       tags: Optional[Dict[str, str]] = None) -> str:
         """
         Create a EMR cluster with instance fleets configuration
         https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-instance-fleet.html
@@ -404,6 +418,10 @@ def create_cluster(self,
         :param spark_glue_catalog: Spark integration with Glue Catalog?
         :param hive_glue_catalog: Hive integration with Glue Catalog?
         :param presto_glue_catalog: Presto integration with Glue Catalog?
+        :param consistent_view: Consistent view allows EMR clusters to check for list and read-after-write consistency for Amazon S3 objects written by or synced with EMRFS. (https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-consistent-view.html)
+        :param consistent_view_retry_seconds: Delay between the tries (seconds)
+        :param consistent_view_retry_count: Number of tries
+        :param consistent_view_table_name: Name of the DynamoDB table to store the consistent view data
         :param bootstraps_paths: Bootstraps paths (e.g ["s3://BUCKET_NAME/script.sh"])
         :param debugging: Debugging enabled?
         :param applications: List of applications (e.g ["Hadoop", "Spark", "Ganglia", "Hive"])

awswrangler/session.py

@@ -1,7 +1,8 @@
-from typing import Optional, Dict
 import os
-from logging import getLogger, Logger
 import importlib
+from typing import Optional, Dict
+from sys import version_info
+from logging import getLogger, Logger
 
 import boto3  # type: ignore
 from botocore.config import Config  # type: ignore
@@ -19,7 +20,7 @@
 from awswrangler.exceptions import AWSCredentialsNotFound
 
 PYSPARK_INSTALLED: bool = False
-if importlib.util.find_spec("pyspark"):  # type: ignore
+if version_info < (3, 8) and importlib.util.find_spec("pyspark"):  # type: ignore
     PYSPARK_INSTALLED = True
     from awswrangler.spark import Spark
 

requirements-dev.txt

@@ -1,3 +1,4 @@
+awscli~=1.17.13
 yapf~=0.29.0
 mypy~=0.761
 flake8~=3.7.9
@@ -6,7 +7,7 @@ scikit-learn~=0.22.1
 cfn-lint~=0.27.4
 twine~=3.1.1
 wheel~=0.34.2
-sphinx~=2.3.1
+sphinx~=2.4.0
 pyspark~=2.4.5
 pyspark-stubs~=2.4.0.post7
 jupyter~=1.0.0

testing/test_awswrangler/test_emr.py

@@ -41,7 +41,7 @@ def test_cluster(session, bucket, cloudformation_outputs):
         steps.append(session.emr.build_step(name=cmd, command=cmd))
     cluster_id = session.emr.create_cluster(cluster_name="wrangler_cluster",
                                             logging_s3_path=f"s3://{bucket}/emr-logs/",
-                                            emr_release="emr-5.27.0",
+                                            emr_release="emr-5.29.0",
                                             subnet_id=cloudformation_outputs["SubnetId"],
                                             emr_ec2_role="EMR_EC2_DefaultRole",
                                             emr_role="EMR_DefaultRole",
@@ -93,7 +93,7 @@ def test_cluster(session, bucket, cloudformation_outputs):
 def test_cluster_single_node(session, bucket, cloudformation_outputs):
     cluster_id = session.emr.create_cluster(cluster_name="wrangler_cluster",
                                             logging_s3_path=f"s3://{bucket}/emr-logs/",
-                                            emr_release="emr-5.27.0",
+                                            emr_release="emr-5.29.0",
                                             subnet_id=cloudformation_outputs["SubnetId"],
                                             emr_ec2_role="EMR_EC2_DefaultRole",
                                             emr_role="EMR_DefaultRole",
@@ -122,6 +122,10 @@ def test_cluster_single_node(session, bucket, cloudformation_outputs):
                                             spark_glue_catalog=False,
                                             hive_glue_catalog=False,
                                             presto_glue_catalog=False,
+                                            consistent_view=True,
+                                            consistent_view_retry_count=6,
+                                            consistent_view_retry_seconds=15,
+                                            consistent_view_table_name="EMRConsistentView",
                                             bootstraps_paths=None,
                                             debugging=False,
                                             applications=["Hadoop", "Spark", "Ganglia", "Hive"],