diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9bed862e52f..a83dd049e92 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -79,6 +79,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#1536](https://github.com/open-telemetry/opentelemetry-python/pull/1536))
 - Fix TraceState to adhere to specs
   ([#1502](https://github.com/open-telemetry/opentelemetry-python/pull/1502))
+- Update Resource `merge` key conflict precedence
+  ([#1544](https://github.com/open-telemetry/opentelemetry-python/pull/1544))
 
 ### Removed
 - `opentelemetry-api` Remove ThreadLocalRuntimeContext since python3.4 is not supported.
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
index 436b4d6c646..2b832cff7d0 100644
--- a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
@@ -155,16 +155,27 @@
 
 
 class Resource:
+    """A Resource is an immutable representation of the entity producing telemetry as Attributes.
+    """
+
     def __init__(self, attributes: Attributes):
         self._attributes = attributes.copy()
 
     @staticmethod
     def create(attributes: typing.Optional[Attributes] = None) -> "Resource":
+        """Creates a new `Resource` from attributes.
+
+        Args:
+            attributes: Optional zero or more key-value pairs.
+
+        Returns:
+            The newly-created Resource.
+        """
         if not attributes:
-            resource = _DEFAULT_RESOURCE
-        else:
-            resource = _DEFAULT_RESOURCE.merge(Resource(attributes))
-        resource = resource.merge(OTELResourceDetector().detect())
+            attributes = {}
+        resource = _DEFAULT_RESOURCE.merge(
+            OTELResourceDetector().detect()
+        ).merge(Resource(attributes))
         if not resource.attributes.get(SERVICE_NAME, None):
             default_service_name = "unknown_service"
             process_executable_name = resource.attributes.get(
@@ -186,11 +197,19 @@ def attributes(self) -> Attributes:
         return self._attributes.copy()
 
     def merge(self, other: "Resource") -> "Resource":
+        """Merges this resource and an updating resource into a new `Resource`.
+
+        If a key exists on both the old and updating resource, the value of the
+        updating resource will override the old resource value.
+
+        Args:
+            other: The other resource to be merged.
+
+        Returns:
+            The newly-created Resource.
+        """
         merged_attributes = self.attributes
-        # pylint: disable=protected-access
-        for key, value in other._attributes.items():
-            if key not in merged_attributes or merged_attributes[key] == "":
-                merged_attributes[key] = value
+        merged_attributes.update(other.attributes)
         return Resource(merged_attributes)
 
     def __eq__(self, other: object) -> bool:
diff --git a/opentelemetry-sdk/tests/resources/test_resources.py b/opentelemetry-sdk/tests/resources/test_resources.py
index 1121459fd00..3effa5d2452 100644
--- a/opentelemetry-sdk/tests/resources/test_resources.py
+++ b/opentelemetry-sdk/tests/resources/test_resources.py
@@ -99,7 +99,7 @@ def test_resource_merge_empty_string(self):
         )
         self.assertEqual(
             left.merge(right),
-            resources.Resource({"service": "ui", "host": "service-host"}),
+            resources.Resource({"service": "not-ui", "host": "service-host"}),
         )
 
     def test_immutability(self):
@@ -154,7 +154,6 @@ def test_aggregated_resources_with_static_resource(self):
             static_resource,
         )
 
-        # Static resource values should never be overwritten
         resource_detector = mock.Mock(spec=resources.ResourceDetector)
         resource_detector.detect.return_value = resources.Resource(
             {"static_key": "try_to_overwrite_existing_value", "key": "value"}
@@ -163,7 +162,12 @@ def test_aggregated_resources_with_static_resource(self):
             resources.get_aggregated_resources(
                 [resource_detector], initial_resource=static_resource
             ),
-            resources.Resource({"static_key": "static_value", "key": "value"}),
+            resources.Resource(
+                {
+                    "static_key": "try_to_overwrite_existing_value",
+                    "key": "value",
+                }
+            ),
         )
 
     def test_aggregated_resources_multiple_detectors(self):
@@ -184,7 +188,6 @@ def test_aggregated_resources_multiple_detectors(self):
             }
         )
 
-        # New values should not overwrite existing values
         self.assertEqual(
             resources.get_aggregated_resources(
                 [resource_detector1, resource_detector2, resource_detector3]
@@ -192,8 +195,8 @@ def test_aggregated_resources_multiple_detectors(self):
             resources.Resource(
                 {
                     "key1": "value1",
-                    "key2": "value2",
-                    "key3": "value3",
+                    "key2": "try_to_overwrite_existing_value",
+                    "key3": "try_to_overwrite_existing_value",
                     "key4": "value4",
                 }
             ),
@@ -216,6 +219,21 @@ def test_resource_detector_raise_error(self):
             Exception, resources.get_aggregated_resources, [resource_detector]
         )
 
+    @mock.patch.dict(
+        os.environ,
+        {"OTEL_RESOURCE_ATTRIBUTES": "key1=env_value1,key2=env_value2"},
+    )
+    def test_env_priority(self):
+        resource_env = resources.Resource.create()
+        self.assertEqual(resource_env.attributes["key1"], "env_value1")
+        self.assertEqual(resource_env.attributes["key2"], "env_value2")
+
+        resource_env_override = resources.Resource.create(
+            {"key1": "value1", "key2": "value2"}
+        )
+        self.assertEqual(resource_env_override.attributes["key1"], "value1")
+        self.assertEqual(resource_env_override.attributes["key2"], "value2")
+
 
 class TestOTELResourceDetector(unittest.TestCase):
     def setUp(self) -> None: