diff --git a/sentry_sdk/integrations/redis/_async_common.py b/sentry_sdk/integrations/redis/_async_common.py
index bd83d22191..956fe91154 100644
--- a/sentry_sdk/integrations/redis/_async_common.py
+++ b/sentry_sdk/integrations/redis/_async_common.py
@@ -2,13 +2,16 @@
 
 import sentry_sdk
 from sentry_sdk.consts import OP, SPANDATA
-from sentry_sdk.integrations.redis.consts import SPAN_ORIGIN
+from sentry_sdk.integrations.redis.consts import (
+    SPAN_ORIGIN,
+)
 from sentry_sdk.integrations.redis.modules.caches import (
     _compile_cache_span_properties,
     _set_cache_data,
 )
 from sentry_sdk.integrations.redis.modules.queries import _compile_db_span_properties
 from sentry_sdk.integrations.redis.utils import (
+    _extract_key,
     _get_safe_command,
     _set_client_data,
     _set_pipeline_data,
@@ -42,6 +45,16 @@ async def _sentry_execute(self: "Any", *args: "Any", **kwargs: "Any") -> "Any":
         if client.get_integration(RedisIntegration) is None:
             return await old_execute(self, *args, **kwargs)
 
+        sentry_sdk.add_breadcrumb(
+            message="redis.pipeline.execute",
+            type="redis",
+            category="redis",
+            data={
+                "redis.is_cluster": is_cluster,
+                "redis.transaction": False if is_cluster else self.is_transaction,
+            },
+        )
+
         span_streaming = has_span_streaming_enabled(client.options)
 
         span: "Union[Span, StreamedSpan]"
@@ -103,6 +116,24 @@ async def _sentry_execute_command(
         if integration is None:
             return await old_execute_command(self, name, *args, **kwargs)
 
+        db_properties = _compile_db_span_properties(integration, name, args)
+
+        breadcrumb_data = {
+            "redis.is_cluster": is_cluster,
+            "redis.command": name,
+            "db.operation": name,
+        }
+        key = _extract_key(name, args)
+        if key is not None:
+            breadcrumb_data["redis.key"] = key
+
+        sentry_sdk.add_breadcrumb(
+            message=db_properties["description"],
+            type="redis",
+            category="redis",
+            data=breadcrumb_data,
+        )
+
         span_streaming = has_span_streaming_enabled(client.options)
 
         if span_streaming and sentry_sdk.traces.get_current_span() is None:
@@ -140,8 +171,6 @@ async def _sentry_execute_command(
                 )
             cache_span.__enter__()
 
-        db_properties = _compile_db_span_properties(integration, name, args)
-
         additional_db_span_attributes = {}
         with capture_internal_exceptions():
             additional_db_span_attributes[SPANDATA.DB_QUERY_TEXT] = _get_safe_command(
diff --git a/sentry_sdk/integrations/redis/_sync_common.py b/sentry_sdk/integrations/redis/_sync_common.py
index 3afa7f282c..fcb1822094 100644
--- a/sentry_sdk/integrations/redis/_sync_common.py
+++ b/sentry_sdk/integrations/redis/_sync_common.py
@@ -2,13 +2,16 @@
 
 import sentry_sdk
 from sentry_sdk.consts import OP, SPANDATA
-from sentry_sdk.integrations.redis.consts import SPAN_ORIGIN
+from sentry_sdk.integrations.redis.consts import (
+    SPAN_ORIGIN,
+)
 from sentry_sdk.integrations.redis.modules.caches import (
     _compile_cache_span_properties,
     _set_cache_data,
 )
 from sentry_sdk.integrations.redis.modules.queries import _compile_db_span_properties
 from sentry_sdk.integrations.redis.utils import (
+    _extract_key,
     _get_safe_command,
     _set_client_data,
     _set_pipeline_data,
@@ -39,8 +42,17 @@ def sentry_patched_execute(self: "Any", *args: "Any", **kwargs: "Any") -> "Any":
         if client.get_integration(RedisIntegration) is None:
             return old_execute(self, *args, **kwargs)
 
-        span_streaming = has_span_streaming_enabled(client.options)
+        sentry_sdk.add_breadcrumb(
+            message="redis.pipeline.execute",
+            type="redis",
+            category="redis",
+            data={
+                "redis.is_cluster": is_cluster,
+                "redis.transaction": False if is_cluster else self.transaction,
+            },
+        )
 
+        span_streaming = has_span_streaming_enabled(client.options)
         span: "Union[Span, StreamedSpan]"
         if span_streaming:
             if sentry_sdk.traces.get_current_span() is None:
@@ -102,6 +114,24 @@ def sentry_patched_execute_command(
         if integration is None:
             return old_execute_command(self, name, *args, **kwargs)
 
+        db_properties = _compile_db_span_properties(integration, name, args)
+
+        breadcrumb_data = {
+            "redis.is_cluster": is_cluster,
+            "redis.command": name,
+            "db.operation": name,
+        }
+        key = _extract_key(name, args)
+        if key is not None:
+            breadcrumb_data["redis.key"] = key
+
+        sentry_sdk.add_breadcrumb(
+            message=db_properties["description"],
+            type="redis",
+            category="redis",
+            data=breadcrumb_data,
+        )
+
         span_streaming = has_span_streaming_enabled(client.options)
 
         if span_streaming and sentry_sdk.traces.get_current_span() is None:
@@ -139,8 +169,6 @@ def sentry_patched_execute_command(
                 )
             cache_span.__enter__()
 
-        db_properties = _compile_db_span_properties(integration, name, args)
-
         additional_db_span_attributes = {}
         with capture_internal_exceptions():
             additional_db_span_attributes[SPANDATA.DB_QUERY_TEXT] = _get_safe_command(
diff --git a/sentry_sdk/integrations/redis/utils.py b/sentry_sdk/integrations/redis/utils.py
index c12752a530..c9cf38cdbd 100644
--- a/sentry_sdk/integrations/redis/utils.py
+++ b/sentry_sdk/integrations/redis/utils.py
@@ -153,12 +153,22 @@ def _set_client_data(
             span.set_tag("redis.command", name)
             span.set_tag(SPANDATA.DB_OPERATION, name)
 
-    if name and args:
-        name_low = name.lower()
-        if (name_low in _SINGLE_KEY_COMMANDS) or (
-            name_low in _MULTI_KEY_COMMANDS and len(args) == 1
-        ):
-            if isinstance(span, StreamedSpan):
-                span.set_attribute("db.redis.key", args[0])
-            else:
-                span.set_tag("redis.key", args[0])
+    key = _extract_key(name, args)
+    if key is not None:
+        if isinstance(span, StreamedSpan):
+            span.set_attribute("db.redis.key", key)
+        else:
+            span.set_tag("redis.key", key)
+
+
+def _extract_key(name: str, args: "Any") -> "Optional[str]":
+    if not name or not args:
+        return None
+
+    name_low = name.lower()
+    if (name_low in _SINGLE_KEY_COMMANDS) or (
+        name_low in _MULTI_KEY_COMMANDS and len(args) == 1
+    ):
+        return args[0]
+
+    return None
diff --git a/sentry_sdk/tracing_utils.py b/sentry_sdk/tracing_utils.py
index 6c903cd21d..b3658f6f91 100644
--- a/sentry_sdk/tracing_utils.py
+++ b/sentry_sdk/tracing_utils.py
@@ -213,12 +213,7 @@ def record_sql_queries(
 def maybe_create_breadcrumbs_from_span(
     scope: "sentry_sdk.Scope", span: "sentry_sdk.tracing.Span"
 ) -> None:
-    if span.op == OP.DB_REDIS:
-        scope.add_breadcrumb(
-            message=span.description, type="redis", category="redis", data=span._tags
-        )
-
-    elif span.op == OP.HTTP_CLIENT:
+    if span.op == OP.HTTP_CLIENT:
         level = None
         status_code = span._data.get(SPANDATA.HTTP_STATUS_CODE)
         if status_code: