diff --git a/CHANGELOG.next.md b/CHANGELOG.next.md
index 3c8577cb6c..c53e8cf33b 100644
--- a/CHANGELOG.next.md
+++ b/CHANGELOG.next.md
@@ -31,6 +31,7 @@ Thanks, you're awesome :-) -->
 * Added ability to supply free-form usage documentation per fieldset. #988
 * Added the `path` key when type is `alias`, to support the [alias field type](https://www.elastic.co/guide/en/elasticsearch/reference/current/alias.html). #877
 * Added support for `scaled_float`'s mandatory parameter `scaling_factor`. #1042
+* Added ability for --oss flag to fall back `constant_keyword` to `keyword`. #1046
 
 #### Improvements
 
diff --git a/scripts/schema/oss.py b/scripts/schema/oss.py
index ba38a254b1..bfc07bb071 100644
--- a/scripts/schema/oss.py
+++ b/scripts/schema/oss.py
@@ -13,6 +13,7 @@
 from schema import visitor
 
 TYPE_FALLBACKS = {
+    'constant_keyword': 'keyword',
     'wildcard': 'keyword',
     'version': 'keyword'
 }
@@ -25,5 +26,6 @@ def fallback(fields):
 
 def perform_fallback(field):
     """Performs a best effort fallback of basic data types to equivalent OSS data types."""
-    if field['field_details']['type'] in TYPE_FALLBACKS.keys():
-        field['field_details']['type'] = TYPE_FALLBACKS[field['field_details']['type']]
+    fallback_type = TYPE_FALLBACKS.get(field['field_details']['type'])
+    if fallback_type:
+        field['field_details']['type'] = fallback_type
diff --git a/scripts/tests/unit/test_schema_oss.py b/scripts/tests/unit/test_schema_oss.py
index 4ac08d9d08..910b7959ca 100644
--- a/scripts/tests/unit/test_schema_oss.py
+++ b/scripts/tests/unit/test_schema_oss.py
@@ -14,6 +14,8 @@ class TestSchemaOss(unittest.TestCase):
     def setUp(self):
         self.maxDiff = None
 
+    # Fallbacks
+
     def test_wildcard_fallback(self):
         field = {'field_details': {'name': 'myfield', 'type': 'wildcard'}}
         oss.perform_fallback(field)
@@ -24,12 +26,19 @@ def test_version_fallback(self):
         oss.perform_fallback(field)
         self.assertEqual('keyword', field['field_details']['type'])
 
+    def test_constant_keyword_fallback(self):
+        field = {'field_details': {'name': 'myfield', 'type': 'constant_keyword'}}
+        oss.perform_fallback(field)
+        self.assertEqual('keyword', field['field_details']['type'])
+
+    # Not falling back
+
     def test_basic_without_fallback(self):
         field = {'field_details': {'name': 'myfield', 'type': 'histogram'}}
         oss.perform_fallback(field)
         self.assertEqual('histogram', field['field_details']['type'])
 
-    def test_oss_no_fallback(self):
+    def test_oss_no_fallback_needed(self):
         field = {'field_details': {'name': 'myfield', 'type': 'keyword'}}
         oss.perform_fallback(field)
         self.assertEqual('keyword', field['field_details']['type'])