diff --git a/CHANGES.md b/CHANGES.md
index e5632b2608fe..b27e315b9906 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -74,6 +74,7 @@
 
 * Support configuring Firestore database on ReadFn transforms (Java) ([#36904](https://github.com/apache/beam/issues/36904)).
 * (Python) Inference args are now allowed in most model handlers, except where they are explicitly/intentionally disallowed ([#37093](https://github.com/apache/beam/issues/37093)).
+* Include logger name in FnApiLogRecordHandler log entries for filtering and debugging (Python) ([#37146](https://github.com/apache/beam/issues/37146)).
 
 ## Breaking Changes
 
diff --git a/sdks/python/apache_beam/runners/worker/log_handler.py b/sdks/python/apache_beam/runners/worker/log_handler.py
index 69815acc7194..5c49c58083db 100644
--- a/sdks/python/apache_beam/runners/worker/log_handler.py
+++ b/sdks/python/apache_beam/runners/worker/log_handler.py
@@ -144,6 +144,10 @@ def emit(self, record: logging.LogRecord) -> None:
           current_state.name_context.transform_id):
         log_entry.transform_id = current_state.name_context.transform_id
 
+    # Include the logger name in custom_data for filtering and debugging.
+    if record.name:
+      log_entry.custom_data.fields['logger'].string_value = record.name
+
     try:
       self._log_entry_queue.put(log_entry, block=False)
     except queue.Full:
diff --git a/sdks/python/apache_beam/runners/worker/log_handler_test.py b/sdks/python/apache_beam/runners/worker/log_handler_test.py
index 7018cfaf683b..3b9cf8a751a1 100644
--- a/sdks/python/apache_beam/runners/worker/log_handler_test.py
+++ b/sdks/python/apache_beam/runners/worker/log_handler_test.py
@@ -233,6 +233,15 @@ def test_context(self):
     finally:
       statesampler.set_current_tracker(None)
 
+  def test_logger_name_in_custom_data(self):
+    """Tests that logger name is included in custom_data."""
+    _LOGGER.info('test message')
+    self.fn_log_handler.close()
+
+    log_entry = self.test_logging_service.log_records_received[0].log_entries[0]
+    self.assertEqual(
+        log_entry.custom_data.fields['logger'].string_value, __name__)
+
   def test_extracts_transform_id_during_exceptions(self):
     """Tests that transform ids are captured during user code exceptions."""
     descriptor = beam_fn_api_pb2.ProcessBundleDescriptor()