[SPARK-58837][CONNECT] Use per-module Python loggers

dev/sparktestsupport/modules.py

@@ -1280,6 +1280,7 @@ def __hash__(self):
         "pyspark.sql.tests.connect.test_connect_column",
         "pyspark.sql.tests.connect.test_connect_creation",
         "pyspark.sql.tests.connect.test_connect_readwriter",
+        "pyspark.sql.tests.connect.test_connect_logging",
         "pyspark.sql.tests.connect.test_connect_retry",
         "pyspark.sql.tests.connect.test_connect_session",
         "pyspark.sql.tests.connect.test_connect_local_server",

python/docs/source/development/logger.rst

@@ -167,3 +167,44 @@ This approach aligns with the standard Python logging practices.
     file_logger.warning(f"User {user} takes an {action}", user=user, action=action)
 
 The log messages will be saved in `application.log` in the same JSON format.
+
+Spark Connect Client Logging
+============================
+
+The Spark Connect Python client logs through one logger per module, all rooted at
+``pyspark.sql.connect``. Every Connect log record is emitted through a single handler on that
+root logger, in the same JSON format described above.
+
+============================================  ==========================================
+Logger                                        Covers
+============================================  ==========================================
+``pyspark.sql.connect``                       Root of the hierarchy, sets the default level
+``pyspark.sql.connect.client.core``           RPC requests and responses, Arrow batches, plan compression
+``pyspark.sql.connect.client.retries``        Retries and backoff, including gRPC ``UNAVAILABLE``
+``pyspark.sql.connect.client.reattach``       Reattachable execution streams
+``pyspark.sql.connect.client.artifact``       Artifact uploads
+``pyspark.sql.connect.dataframe``             DataFrame operations
+``pyspark.sql.connect.plan``                  Logical plan construction and cached relation cleanup
+``pyspark.sql.connect.session``               Session lifecycle
+============================================  ==========================================
+
+Connect logging is disabled by default. Setting the ``SPARK_CONNECT_LOG_LEVEL`` environment
+variable enables the whole hierarchy at that level:
+
+.. code-block:: bash
+
+    export SPARK_CONNECT_LOG_LEVEL=debug
+
+``DEBUG`` on the whole hierarchy is verbose, because ``client.core`` dumps every request and
+response. To follow one area instead, set the level on that logger with the standard Python
+logging module. For example, to watch the client retry against an unresponsive server without
+the rest of the output:
+
+.. code-block:: python
+
+    import logging
+
+    logging.getLogger("pyspark.sql.connect.client.retries").setLevel(logging.DEBUG)
+
+Levels are read when a message is logged, so this takes effect at any point, including after the
+session has been created.

python/pyspark/logger/logger.py

@@ -139,21 +139,7 @@ class PySparkLogger(logging.Logger):
     """
 
     def __init__(self, name: str = "PySparkLogger"):
-        from pyspark.logger.worker_io import JSONFormatterWithMarker
-
-        super().__init__(name, level=logging.WARN)
-
-        root_logger = logging.getLogger()
-        if any(
-            isinstance(h, logging.StreamHandler)
-            and isinstance(h.formatter, JSONFormatterWithMarker)
-            for h in root_logger.handlers
-        ):
-            # Likely in the `capture_outputs` context, so don't add a handler
-            return
-
-        _handler = logging.StreamHandler()
-        self.addHandler(_handler)
+        super().__init__(name)
 
     def addHandler(self, handler: logging.Handler) -> None:
         """

python/pyspark/sql/connect/client/artifact.py

@@ -33,7 +33,9 @@
 import pyspark.sql.connect.proto as proto
 import pyspark.sql.connect.proto.base_pb2_grpc as grpc_lib
 from pyspark.errors import PySparkRuntimeError, PySparkValueError
-from pyspark.sql.connect.logging import logger
+from pyspark.sql.connect.logging import getLogger
+
+logger = getLogger(__name__)
 
 JAR_PREFIX: str = "jars"
 PYFILE_PREFIX: str = "pyfiles"

python/pyspark/sql/connect/client/core.py

@@ -96,7 +96,7 @@
     LiteralExpression,
     PythonUDF,
 )
-from pyspark.sql.connect.logging import logger
+from pyspark.sql.connect.logging import getLogger
 from pyspark.sql.connect.observation import Observation
 from pyspark.sql.connect.plan import (
     CommonInlineUserDefinedDataSource,
@@ -126,6 +126,8 @@
     from pyspark.sql.datasource import DataSource
 
 
+logger = getLogger(__name__)
+
 PYSPARK_ROOT = os.path.dirname(pyspark.__file__)
 _OPERATION_ID_METADATA_KEY = "spark-connect-operation-id"
 

python/pyspark/sql/connect/client/reattach.py

@@ -29,9 +29,11 @@
 import pyspark.sql.connect.proto.base_pb2_grpc as grpc_lib
 from pyspark.errors import PySparkRuntimeError
 from pyspark.sql.connect.client.retries import RetryException, Retrying
-from pyspark.sql.connect.logging import logger
+from pyspark.sql.connect.logging import getLogger
 from pyspark.util import disable_gc
 
+logger = getLogger(__name__)
+
 
 class ExecutePlanResponseReattachableIterator(Generator):
     """

python/pyspark/sql/connect/client/retries.py

@@ -27,7 +27,9 @@
 from grpc_status import rpc_status
 
 from pyspark.errors import PySparkRuntimeError
-from pyspark.sql.connect.logging import logger
+from pyspark.sql.connect.logging import getLogger
+
+logger = getLogger(__name__)
 
 """
 This module contains retry system. The system is designed to be

python/pyspark/sql/connect/dataframe.py

@@ -71,7 +71,7 @@
 )
 from pyspark.sql.connect.functions import builtin as F
 from pyspark.sql.connect.group import GroupedData
-from pyspark.sql.connect.logging import logger
+from pyspark.sql.connect.logging import getLogger
 from pyspark.sql.connect.merge import MergeIntoWriter
 from pyspark.sql.connect.readwriter import DataFrameWriter, DataFrameWriterV2
 from pyspark.sql.connect.streaming.readwriter import DataStreamWriter
@@ -111,6 +111,9 @@
     from pyspark.sql.plot import PySparkPlotAccessor
 
 
+logger = getLogger(__name__)
+
+
 class DataFrame(ParentDataFrame):
     def __new__(
         cls,

python/pyspark/sql/connect/logging.py

@@ -20,9 +20,60 @@
 import os
 from typing import Optional
 
-from pyspark.logger import PySparkLogger
+from pyspark.logger.logger import JSONFormatter
 
-__all__ = ["configureLogging", "getLogLevel"]
+__all__ = ["configureLogging", "getLogger", "getLogLevel"]
+
+# Root of the Spark Connect logger hierarchy. It owns the handler that all Connect loggers
+# write through, and its level is the default for every logger below it.
+_CONNECT_LOGGER_NAME = "pyspark.sql.connect"
+
+# Above CRITICAL, so nothing is emitted. Used instead of `Logger.disabled` because `disabled`
+# is only honored on the logger a message is logged on, which would leave child loggers
+# inheriting the standard library root logger's WARNING instead of staying silent.
+_LOG_LEVEL_OFF = logging.CRITICAL + 1
+
+# The one handler for the whole hierarchy. Child loggers propagate their records up to it.
+_HANDLER = logging.StreamHandler()
+_HANDLER.setFormatter(JSONFormatter())
+logging.getLogger(_CONNECT_LOGGER_NAME).addHandler(_HANDLER)
+
+
+def getLogger(name: Optional[str] = None) -> logging.Logger:
+    """
+    Return a logger in the Spark Connect logger hierarchy, creating it if necessary.
+
+    Spark Connect logs through one logger per module, all rooted at ``pyspark.sql.connect``,
+    so a single area can be turned on without enabling the rest:
+
+    .. code-block:: python
+
+        import logging
+
+        logging.getLogger("pyspark.sql.connect.client.retries").setLevel(logging.DEBUG)
+
+    The available loggers are ``pyspark.sql.connect`` and, below it, ``client.core``,
+    ``client.retries``, ``client.reattach``, ``client.artifact``, ``dataframe``, ``plan``,
+    and ``session``.
+
+    Parameters
+    ----------
+    name : str, optional
+        Name of the logger, either relative to ``pyspark.sql.connect`` or fully qualified.
+        When omitted, the root Spark Connect logger is returned.
+
+    .. versionadded:: 4.4.0
+    """
+    if not name or name == _CONNECT_LOGGER_NAME:
+        qualified_name = _CONNECT_LOGGER_NAME
+    elif name.startswith(_CONNECT_LOGGER_NAME + "."):
+        qualified_name = name
+    else:
+        qualified_name = f"{_CONNECT_LOGGER_NAME}.{name}"
+
+    # Deliberately a plain logger rather than PySparkLogger, which attaches a handler to every
+    # instance it creates. Records propagate to the handler on the root Connect logger instead.
+    return logging.getLogger(qualified_name)
 
 
 def configureLogging(level: Optional[str] = None) -> logging.Logger:
@@ -32,25 +83,30 @@ def configureLogging(level: Optional[str] = None) -> logging.Logger:
     the SPARK_CONNECT_LOG_LEVEL environment variable.
     When both are absent, logging is disabled.
 
+    The level applies to the root Spark Connect logger, and therefore to every Connect logger
+    that does not have a level of its own. See :func:`getLogger` for enabling a single logger.
+
     .. versionadded:: 4.0.0
+
+    .. versionchanged:: 4.4.0
+        Repeated calls no longer attach an additional handler, and a level set on the root
+        Spark Connect logger before PySpark is imported is no longer overwritten.
     """
-    logger = PySparkLogger.getLogger(__name__)
-    handler = logging.StreamHandler()
-    handler.setFormatter(
-        logging.Formatter(fmt="%(asctime)s %(process)d %(levelname)s %(funcName)s %(message)s")
-    )
-    logger.addHandler(handler)
+    logger = logging.getLogger(_CONNECT_LOGGER_NAME)
+
+    if level is None:
+        level = os.environ.get("SPARK_CONNECT_LOG_LEVEL")
 
     if level is not None:
         logger.setLevel(level.upper())
-    elif "SPARK_CONNECT_LOG_LEVEL" in os.environ:
-        logger.setLevel(os.environ["SPARK_CONNECT_LOG_LEVEL"].upper())
-    else:
-        logger.disabled = True
+        logger.disabled = False
+    elif logger.level == logging.NOTSET:
+        logger.setLevel(_LOG_LEVEL_OFF)
     return logger
 
 
-# Instantiate the logger based on the environment configuration.
+# Instantiate the root Spark Connect logger based on the environment configuration. Kept as a
+# module-level name for backwards compatibility, new code should use getLogger(__name__).
 logger = configureLogging()
 
 
@@ -63,6 +119,6 @@ def getLogLevel() -> Optional[int]:
     .. versionadded:: 3.5.0
     """
 
-    if not logger.disabled:
+    if not logger.disabled and logger.level < _LOG_LEVEL_OFF:
         return logger.level
     return None

python/pyspark/sql/connect/plan.py

@@ -50,7 +50,7 @@
 from pyspark.sql.column import Column
 from pyspark.sql.connect.conversion import storage_level_to_proto
 from pyspark.sql.connect.expressions import Expression, SubqueryExpression
-from pyspark.sql.connect.logging import logger
+from pyspark.sql.connect.logging import getLogger
 from pyspark.sql.connect.proto import base_pb2 as spark_dot_connect_dot_base__pb2
 from pyspark.sql.connect.types import UnparsedDataType, pyspark_types_to_proto_types
 from pyspark.sql.types import DataType, StructType
@@ -63,6 +63,9 @@
     from pyspark.sql.connect.udf import UserDefinedFunction
 
 
+logger = getLogger(__name__)
+
+
 class LogicalPlan:
     _lock: Lock = Lock()
     _nextPlanId: int = 0

python/pyspark/sql/connect/session.py

@@ -61,7 +61,7 @@
 from pyspark.sql.connect.conf import RuntimeConf
 from pyspark.sql.connect.dataframe import DataFrame
 from pyspark.sql.connect.functions import builtin as F
-from pyspark.sql.connect.logging import logger
+from pyspark.sql.connect.logging import getLogger
 from pyspark.sql.connect.plan import (
     SQL,
     CachedRelation,
@@ -114,6 +114,9 @@
     from pyspark.sql.connect.udtf import UDTFRegistration
 
 
+logger = getLogger(__name__)
+
+
 class SparkSession:
     # The active SparkSession for the current thread
     _active_session: ClassVar[threading.local] = threading.local()