diff --git a/bundle/config/mutator/validate_unique_resource_keys.go b/bundle/config/mutator/validate_unique_resource_keys.go
new file mode 100644
index 00000000..6698b822
--- /dev/null
+++ b/bundle/config/mutator/validate_unique_resource_keys.go
@@ -0,0 +1,72 @@
+package mutator
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
+)
+
+// TODO: Ensure the solution here works for dup keys in the same resource type.
+
+type validateUniqueResourceKeys struct{}
+
+func ValidateUniqueResourceKeys() bundle.Mutator {
+	return &validateUniqueResourceKeys{}
+}
+
+func (m *validateUniqueResourceKeys) Name() string {
+	return "ValidateUniqueResourceKeys"
+}
+
+// TODO: Ensure all duplicate key sites are returned to the user in the diagnostics.
+func (m *validateUniqueResourceKeys) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
+	diags := diag.Diagnostics{}
+
+	err := b.Config.Mutate(func(v dyn.Value) (dyn.Value, error) {
+		paths := make(map[string]dyn.Path)
+		rv := v.Get("resources")
+
+		// Walk the resources tree and accumulate the resource identifiers.
+		return dyn.Walk(rv, func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
+			// The path is expected to be of length 2, and of the form <resource_type>.<resource_identifier>.
+			// Eg: jobs.my_job, pipelines.my_pipeline, etc.
+			if len(p) < 2 {
+				return v, nil
+			}
+			if len(p) > 2 {
+				return v, dyn.ErrSkip
+			}
+
+			// If the resource identifier already exists in the map, return an error.
+			k := p[1].Key()
+			if _, ok := paths[k]; ok {
+				// Location of the existing resource in the map.
+				ov, _ := dyn.GetByPath(rv, paths[k])
+				ol := ov.Location()
+
+				// Location of the newly encountered with a duplicate name.
+				nv, _ := dyn.GetByPath(rv, p)
+				nl := nv.Location()
+
+				// Error, encountered a duplicate resource identifier.
+				// TODO: Set location for one of the resources?
+				diags = append(diags, diag.Diagnostic{
+					Severity: diag.Error,
+					Summary:  fmt.Sprintf("multiple resources named %s (%s at %s, %s at %s)", k, paths[k].String(), nl, p.String(), ol),
+				})
+			}
+
+			// Accumulate the resource identifier and its path.
+			paths[k] = p
+			return v, nil
+		})
+	})
+	if err != nil {
+		diags = append(diags, diag.FromErr(err)...)
+	}
+
+	return diags
+}
diff --git a/bundle/phases/initialize.go b/bundle/phases/initialize.go
index ded2e198..d15c2323 100644
--- a/bundle/phases/initialize.go
+++ b/bundle/phases/initialize.go
@@ -7,7 +7,6 @@ import (
 	"github.com/databricks/cli/bundle/deploy/metadata"
 	"github.com/databricks/cli/bundle/deploy/terraform"
 	"github.com/databricks/cli/bundle/permissions"
-	"github.com/databricks/cli/bundle/python"
 	"github.com/databricks/cli/bundle/scripts"
 )
 
@@ -18,6 +17,7 @@ func Initialize() bundle.Mutator {
 	return newPhase(
 		"initialize",
 		[]bundle.Mutator{
+			mutator.ValidateUniqueResourceKeys(),
 			mutator.RewriteSyncPaths(),
 			mutator.MergeJobClusters(),
 			mutator.MergeJobTasks(),
@@ -41,7 +41,7 @@ func Initialize() bundle.Mutator {
 			mutator.DefaultQueueing(),
 			mutator.ExpandPipelineGlobPaths(),
 			mutator.TranslatePaths(),
-			python.WrapperWarning(),
+			// python.WrapperWarning(),
 			permissions.ApplyBundlePermissions(),
 			permissions.FilterCurrentUser(),
 			metadata.AnnotateJobs(),