Skip to content

[SPARK-59007][PYTHON][TESTS] Fix flaky Arrow grouped map worker logging tests #58289

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Yicong-Huang wants to merge 1 commit into from
+46 −27
Closed

[SPARK-59007][PYTHON][TESTS] Fix flaky Arrow grouped map worker logging tests #58289

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 46 additions & 27 deletions python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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):
Expand All @@ -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):
Expand Down