diff --git a/python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py b/python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py
index 84fdf0f0a766a..8af706d8a9c50 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py
@@ -27,7 +27,12 @@
 from pyspark.sql.functions import array, col, explode, lit, mean, stddev
 from pyspark.sql.window import Window
 from pyspark.testing.sqlutils import ReusedSQLTestCase
-from pyspark.testing.utils import assertDataFrameEqual, have_pyarrow, pyarrow_requirement_message
+from pyspark.testing.utils import (
+    assertDataFrameEqual,
+    eventually,
+    have_pyarrow,
+    pyarrow_requirement_message,
+)
 from pyspark.util import is_remote_only
 
 if have_pyarrow:
@@ -472,20 +477,27 @@ def func_with_logging(group):
                 df,
             )
 
-            logs = self.spark.tvf.python_worker_logs()
+            # Worker logs are captured asynchronously from the worker's stdout and only
+            # become visible once the trailing block is flushed, so they may not all be
+            # present immediately after the query completes. Poll until they show up.
+            @eventually(timeout=5, catch_assertions=True)
+            def check_logs():
+                logs = self.spark.tvf.python_worker_logs()
+
+                assertDataFrameEqual(
+                    logs.select("level", "msg", "context", "logger"),
+                    [
+                        Row(
+                            level="WARNING",
+                            msg=f"arrow grouped map: {dict(id=lst, value=[v * 10 for v in lst])}",
+                            context={"func_name": func_with_logging.__name__},
+                            logger="test_arrow_grouped_map",
+                        )
+                        for lst in [[0, 2, 4, 6, 8], [1, 3, 5, 7]]
+                    ],
+                )
 
-            assertDataFrameEqual(
-                logs.select("level", "msg", "context", "logger"),
-                [
-                    Row(
-                        level="WARNING",
-                        msg=f"arrow grouped map: {dict(id=lst, value=[v * 10 for v in lst])}",
-                        context={"func_name": func_with_logging.__name__},
-                        logger="test_arrow_grouped_map",
-                    )
-                    for lst in [[0, 2, 4, 6, 8], [1, 3, 5, 7]]
-                ],
-            )
+            check_logs()
 
     @unittest.skipIf(is_remote_only(), "Requires JVM access")
     def test_apply_in_arrow_iter_with_logging(self):
@@ -512,20 +524,27 @@ def func_with_logging(group: Iterator[pa.RecordBatch]) -> Iterator[pa.RecordBatc
                 df,
             )
 
-            logs = self.spark.tvf.python_worker_logs()
+            # Worker logs are captured asynchronously from the worker's stdout and only
+            # become visible once the trailing block is flushed, so they may not all be
+            # present immediately after the query completes. Poll until they show up.
+            @eventually(timeout=5, catch_assertions=True)
+            def check_logs():
+                logs = self.spark.tvf.python_worker_logs()
+
+                assertDataFrameEqual(
+                    logs.select("level", "msg", "context", "logger"),
+                    [
+                        Row(
+                            level="WARNING",
+                            msg=f"arrow grouped map: {dict(id=lst, value=[v * 10 for v in lst])}",
+                            context={"func_name": func_with_logging.__name__},
+                            logger="test_arrow_grouped_map",
+                        )
+                        for lst in [[0, 2, 4], [6, 8], [1, 3, 5], [7]]
+                    ],
+                )
 
-            assertDataFrameEqual(
-                logs.select("level", "msg", "context", "logger"),
-                [
-                    Row(
-                        level="WARNING",
-                        msg=f"arrow grouped map: {dict(id=lst, value=[v * 10 for v in lst])}",
-                        context={"func_name": func_with_logging.__name__},
-                        logger="test_arrow_grouped_map",
-                    )
-                    for lst in [[0, 2, 4], [6, 8], [1, 3, 5], [7]]
-                ],
-            )
+            check_logs()
 
 
 class ApplyInArrowTests(ApplyInArrowTestsMixin, ReusedSQLTestCase):