diff --git a/lib/ansible/modules/cloud/amazon/redshift.py b/lib/ansible/modules/cloud/amazon/redshift.py
index b09a589cbb..585aa58a95 100644
--- a/lib/ansible/modules/cloud/amazon/redshift.py
+++ b/lib/ansible/modules/cloud/amazon/redshift.py
@@ -80,6 +80,19 @@ options:
       - VPC security group
     aliases: ['vpc_security_groups']
     default: null
+  skip_final_cluster_snapshot:
+    description:
+      - skip a final snapshot before deleting the cluster. Used only when command=delete.
+    aliases: ['skip_final_snapshot']
+    default: false
+    version_added: "2.4"
+  final_cluster_snapshot_identifier:
+    description:
+      - identifier of the final snapshot to be created before deleting the cluster. If this parameter is provided,
+        final_cluster_snapshot_identifier must be false. Used only when command=delete.
+    aliases: ['final_snapshot_id']
+    default: null
+    version_added: "2.4"
   preferred_maintenance_window:
     description:
       - maintenance window
@@ -149,6 +162,13 @@ EXAMPLES = '''
     identifier=new_cluster
     username=cluster_admin
     password=1nsecure
+
+# Cluster delete example
+- redshift:
+    command: delete
+    identifier: new_cluster
+    skip_final_cluster_snapshot: true
+    wait: true
 '''
 
 RETURN = '''
@@ -327,12 +347,18 @@ def delete_cluster(module, redshift):
     redshift: authenticated redshift connection object
     """
 
-    identifier   = module.params.get('identifier')
-    wait         = module.params.get('wait')
+    identifier = module.params.get('identifier')
+    wait = module.params.get('wait')
     wait_timeout = module.params.get('wait_timeout')
+    skip_final_cluster_snapshot = module.params.get('skip_final_cluster_snapshot')
+    final_cluster_snapshot_identifier = module.params.get('final_cluster_snapshot_identifier')
 
     try:
-        redshift.delete_custer( identifier )
+        redshift.delete_cluster(
+            identifier,
+            skip_final_cluster_snapshot,
+            final_cluster_snapshot_identifier
+        )
     except boto.exception.JSONResponseError as e:
         module.fail_json(msg=str(e))
 
@@ -422,6 +448,8 @@ def main():
         cluster_type                        = dict(choices=['multi-node', 'single-node', ], default='single-node'),
         cluster_security_groups             = dict(aliases=['security_groups'], type='list'),
         vpc_security_group_ids              = dict(aliases=['vpc_security_groups'], type='list'),
+        skip_final_cluster_snapshot         = dict(aliases=['skip_final_snapshot'], type='bool', default=False),
+        final_cluster_snapshot_identifier   = dict(aliases=['final_snapshot_id'], required=False),
         cluster_subnet_group_name           = dict(aliases=['subnet']),
         availability_zone                   = dict(aliases=['aws_zone', 'zone']),
         preferred_maintenance_window        = dict(aliases=['maintance_window', 'maint_window']),
@@ -437,11 +465,16 @@ def main():
         new_cluster_identifier              = dict(aliases=['new_identifier']),
         wait                                = dict(type='bool', default=False),
         wait_timeout                        = dict(type='int', default=300),
-    )
-    )
+    ))
+
+    required_if = [
+        ('command', 'delete', ['skip_final_cluster_snapshot']),
+        ('skip_final_cluster_snapshot', False, ['final_cluster_snapshot_identifier'])
+    ]
 
     module = AnsibleModule(
         argument_spec=argument_spec,
+        required_if=required_if
     )
 
     if not HAS_BOTO: