diff --git a/metaflow/plugins/aws/batch/batch.py b/metaflow/plugins/aws/batch/batch.py
index 1222641b7b3..83b7ca4c722 100644
--- a/metaflow/plugins/aws/batch/batch.py
+++ b/metaflow/plugins/aws/batch/batch.py
@@ -32,6 +32,7 @@
 )
 from metaflow.plugins.aws.aws_utils import sanitize_batch_tag
 from metaflow.plugins.datatools.s3.s3tail import S3Tail
+from metaflow.tagging_util import validate_tags, validate_aws_tag
 
 from .batch_client import BatchClient
 
@@ -347,14 +348,21 @@ def create_job(
                             'value': key_value[1]
                             })
                 for tag in aws_tags_list:
+                    validate_aws_tag(tag)
                     job.tag(tag['key'], tag['value'])
 
-            # add custom tags last to allow override of defaults
             if tags is not None:
                 if not isinstance(tags, dict):
                     raise BatchException("tags must be a dictionary of key-value tags.")
-                for name, value in tags.items():
-                    job.tag(name, value)
+                decorator_aws_tags_list = []
+                for key, val in tags:
+                    aws_tags_list.append({
+                        'key': key,
+                        'value': val
+                        })
+                for tag in decorator_aws_tags_list:
+                    validate_aws_tag(tag)
+                    job.tag(tag['key'], tag['value'])
 
         return job
 
diff --git a/metaflow/tagging_util.py b/metaflow/tagging_util.py
index e124a6e24e1..814bec2367d 100644
--- a/metaflow/tagging_util.py
+++ b/metaflow/tagging_util.py
@@ -1,5 +1,6 @@
 from metaflow.exception import MetaflowTaggingError
 from metaflow.util import unicode_type, bytes_type
+import re
 
 
 def is_utf8_encodable(x):
@@ -29,6 +30,12 @@ def is_utf8_decodable(x):
 # How long may an individual tag value be
 MAX_TAG_SIZE = 500
 
+def validate_aws_tag(tag):
+    validate_tags([tag['key'], tag['value']])
+    if not re.match(r'^[\w\s._-]+$', tag['key']):
+        raise MetaflowTaggingError("Tags must match pattern: ^[\w\s._-]+$")
+    if not re.match(r'^[\w\s._-]*$', tag['value']):
+        raise MetaflowTaggingError("Tags must match pattern: ^[\w\s._-]+$")
 
 def validate_tags(tags, existing_tags=None):
     """