diff --git a/CHANGELOG.next.md b/CHANGELOG.next.md
index 39d6bcc56c..42ea7292eb 100644
--- a/CHANGELOG.next.md
+++ b/CHANGELOG.next.md
@@ -65,6 +65,7 @@ Thanks, you're awesome :-) -->
 * Added support for marking fields, field sets, or field reuse as beta in the documentation. #1051
 * Added support for `constant_keyword`'s optional parameter `value`. #1112
 * Added component templates for ECS field sets. #1156, #1186, #1191
+* Added functionality for merging custom and core multi-fields. #982
 
 #### Improvements
 
diff --git a/scripts/schema/loader.py b/scripts/schema/loader.py
index 07477551af..04f3218ae4 100644
--- a/scripts/schema/loader.py
+++ b/scripts/schema/loader.py
@@ -186,6 +186,28 @@ def nest_fields(field_array):
     return schema_root
 
 
+def array_of_maps_to_map(array_vals):
+    ret_map = {}
+    for map_val in array_vals:
+        name = map_val['name']
+        # if multiple name fields exist in the same custom definition this will take the last one
+        ret_map[name] = map_val
+    return ret_map
+
+
+def map_of_maps_to_array(map_vals):
+    ret_list = []
+    for key in map_vals:
+        ret_list.append(map_vals[key])
+    return sorted(ret_list, key=lambda k: k['name'])
+
+
+def dedup_and_merge_lists(list_a, list_b):
+    list_a_map = array_of_maps_to_map(list_a)
+    list_a_map.update(array_of_maps_to_map(list_b))
+    return map_of_maps_to_array(list_a_map)
+
+
 def merge_fields(a, b):
     """Merge ECS field sets with custom field sets."""
     a = copy.deepcopy(a)
@@ -199,6 +221,14 @@ def merge_fields(a, b):
             a[key].setdefault('field_details', {})
             a[key]['field_details'].setdefault('normalize', [])
             a[key]['field_details']['normalize'].extend(b[key]['field_details'].pop('normalize'))
+        if 'multi_fields' in b[key]['field_details']:
+            a[key].setdefault('field_details', {})
+            a[key]['field_details'].setdefault('multi_fields', [])
+            a[key]['field_details']['multi_fields'] = dedup_and_merge_lists(
+                a[key]['field_details']['multi_fields'], b[key]['field_details']['multi_fields'])
+            # if we don't do this then the update call below will overwrite a's field_details, with the original
+            # contents of b, which undoes our merging the multi_fields
+            del b[key]['field_details']['multi_fields']
         a[key]['field_details'].update(b[key]['field_details'])
         # merge schema details
         if 'schema_details' in b[key]:
diff --git a/scripts/tests/unit/test_schema_loader.py b/scripts/tests/unit/test_schema_loader.py
index de3a718bd5..fde33e0a1c 100644
--- a/scripts/tests/unit/test_schema_loader.py
+++ b/scripts/tests/unit/test_schema_loader.py
@@ -646,6 +646,81 @@ def test_merge_non_array_attributes(self):
         }
         self.assertEqual(merged_fields, expected_fields)
 
+    def test_merge_and_overwrite_multi_fields(self):
+        originalSchema = {
+            'overwrite_field': {
+                'field_details': {
+                    'multi_fields': [
+                        {
+                            'type': 'text',
+                            'name': 'text',
+                            'norms': True
+                        }
+                    ]
+                },
+                'fields': {
+                    'message': {
+                        'field_details': {
+                            'multi_fields': [
+                                {
+                                    'type': 'text',
+                                    'name': 'text'
+                                }
+                            ]
+                        }
+                    }
+                }
+            }
+        }
+
+        customSchema = {
+            'overwrite_field': {
+                'field_details': {
+                    'multi_fields': [
+                        # this entry will completely overwrite the originalSchema's name text entry
+                        {
+                            'type': 'text',
+                            'name': 'text'
+                        }
+                    ]
+                },
+                'fields': {
+                    'message': {
+                        'field_details': {
+                            'multi_fields': [
+                                # this entry will be merged with the originalSchema's multi_fields entries
+                                {
+                                    'type': 'keyword',
+                                    'name': 'a_field'
+                                }
+                            ]
+                        }
+                    }
+                }
+            }
+        }
+        merged_fields = loader.merge_fields(originalSchema, customSchema)
+        expected_overwrite_field_mf = [
+            {
+                'type': 'text',
+                'name': 'text'
+            }
+        ]
+
+        expected_message_mf = [
+            {
+                'type': 'keyword',
+                'name': 'a_field'
+            },
+            {
+                'type': 'text',
+                'name': 'text'
+            }
+        ]
+        self.assertEqual(merged_fields['overwrite_field']['field_details']['multi_fields'], expected_overwrite_field_mf)
+        self.assertEqual(merged_fields['overwrite_field']['fields']['message']['field_details']
+                         ['multi_fields'], expected_message_mf)
+
 
 if __name__ == '__main__':
     unittest.main()