You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/11/12 22:20:44 UTC
(spark) branch master updated: [SPARK-54323][PYTHON] Change the way to access logs to TVF instead of system view
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new fb8dbd7071b1 [SPARK-54323][PYTHON] Change the way to access logs to TVF instead of system view
fb8dbd7071b1 is described below
commit fb8dbd7071b13549d672cc204cec7ab12e7dc520
Author: Takuya Ueshin <ue...@databricks.com>
AuthorDate: Wed Nov 12 14:20:27 2025 -0800
[SPARK-54323][PYTHON] Change the way to access logs to TVF instead of system view
### What changes were proposed in this pull request?
Changes the way to access logs to TVF instead of system view.
```sql
SELECT * FROM python_worker_logs()
```
```py
spark.tvf.python_worker_logs()
```
Also blocks the TVF when the python worker logging is disabled.
```py
>>> spark.conf.get('spark.sql.pyspark.worker.logging.enabled')
'false'
>>> spark.tvf.python_worker_logs().show()
Traceback (most recent call last):
...
pyspark.errors.exceptions.captured.AnalysisException: [FEATURE_NOT_ENABLED] The feature Python Worker Logging is not enabled. Consider setting the config spark.sql.pyspark.worker.logging.enabled to true to enable this capability. SQLSTATE: 56038
>>> spark.conf.set('spark.sql.pyspark.worker.logging.enabled', True)
>>> spark.tvf.python_worker_logs().show()
+---+-----+---+-------+---------+------+
| ts|level|msg|context|exception|logger|
+---+-----+---+-------+---------+------+
+---+-----+---+-------+---------+------+
```
### Why are the changes needed?
There may be namespace conflicts with the other system tables/views, etc..
For example, the variables in SQL has the same namespace `system.session.varname`, which may potentially cause an issue.
### Does this PR introduce _any_ user-facing change?
Yes, the way to access python worker logs will be changed.
### How was this patch tested?
Modified the related tests.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #53021 from ueshin/issues/SPARK-54323/tvf.
Authored-by: Takuya Ueshin <ue...@databricks.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
project/MimaExcludes.scala | 5 +-
python/docs/source/user_guide/bugbusting.ipynb | 12 +-
python/pyspark/sql/connect/tvf.py | 5 +
.../sql/tests/arrow/test_arrow_cogrouped_map.py | 28 +-
.../sql/tests/arrow/test_arrow_grouped_map.py | 54 ++--
python/pyspark/sql/tests/arrow/test_arrow_map.py | 10 +-
.../sql/tests/arrow/test_arrow_udf_grouped_agg.py | 26 +-
.../sql/tests/arrow/test_arrow_udf_scalar.py | 52 ++--
.../sql/tests/arrow/test_arrow_udf_window.py | 28 +-
python/pyspark/sql/tests/arrow/test_arrow_udtf.py | 26 +-
.../sql/tests/pandas/test_pandas_cogrouped_map.py | 28 +-
.../sql/tests/pandas/test_pandas_grouped_map.py | 26 +-
python/pyspark/sql/tests/pandas/test_pandas_map.py | 28 +-
.../tests/pandas/test_pandas_udf_grouped_agg.py | 26 +-
.../sql/tests/pandas/test_pandas_udf_scalar.py | 52 ++--
.../sql/tests/pandas/test_pandas_udf_window.py | 28 +-
python/pyspark/sql/tests/test_python_datasource.py | 298 ++++++++++-----------
python/pyspark/sql/tests/test_udf.py | 142 +++++-----
python/pyspark/sql/tests/test_udtf.py | 106 ++++----
python/pyspark/sql/tvf.py | 39 +++
.../org/apache/spark/sql/TableValuedFunction.scala | 8 +
.../sql/catalyst/analysis/FunctionRegistry.scala | 5 +-
.../sql/catalyst/analysis/RelationResolution.scala | 20 +-
.../plans/logical/pythonLogicalOperators.scala | 64 +++--
.../spark/sql/errors/QueryCompilationErrors.scala | 11 +
.../spark/sql/connect/TableValuedFunction.scala | 4 +
.../spark/sql/classic/TableValuedFunction.scala | 4 +
.../execution/python/PythonWorkerLogsSuite.scala | 30 ++-
28 files changed, 642 insertions(+), 523 deletions(-)
diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index bc8b58847597..b7239c4c6160 100644
--- a/project/MimaExcludes.scala
+++ b/project/MimaExcludes.scala
@@ -52,7 +52,10 @@ object MimaExcludes {
ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.Dataset.repartitionById"),
// [SPARK-54001][CONNECT] Replace block copying with ref-counting in ArtifactManager cloning
- ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.artifact.ArtifactManager.cachedBlockIdList")
+ ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.artifact.ArtifactManager.cachedBlockIdList"),
+
+ // [SPARK-54323][PYTHON] Change the way to access logs to TVF instead of system view
+ ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.TableValuedFunction.python_worker_logs")
)
// Default exclude rules
diff --git a/python/docs/source/user_guide/bugbusting.ipynb b/python/docs/source/user_guide/bugbusting.ipynb
index 75150596600e..407bcbf18e9b 100644
--- a/python/docs/source/user_guide/bugbusting.ipynb
+++ b/python/docs/source/user_guide/bugbusting.ipynb
@@ -945,7 +945,7 @@
"metadata": {},
"outputs": [],
"source": [
- "logs = spark.table(\"system.session.python_worker_logs\")"
+ "logs = spark.tvf.python_worker_logs()"
]
},
{
@@ -1021,7 +1021,7 @@
"df.select(my_udf(\"text\")).show()\n",
"\n",
"# Query the logs\n",
- "logs = spark.table(\"system.session.python_worker_logs\")\n",
+ "logs = spark.tvf.python_worker_logs()\n",
"logs.select(\"level\", \"msg\", \"logger\", \"context\").show(truncate=False)"
]
},
@@ -1076,7 +1076,7 @@
"spark.conf.set(\"spark.sql.pyspark.worker.logging.enabled\", \"true\")\n",
"spark.range(1).select(contextual_udf(lit(\"test\"))).show()\n",
"\n",
- "logs = spark.table(\"system.session.python_worker_logs\")\n",
+ "logs = spark.tvf.python_worker_logs()\n",
"logs.filter(\"logger = 'contextual'\").select(\"msg\", \"context\").show(truncate=False)"
]
},
@@ -1135,7 +1135,7 @@
"spark.conf.set(\"spark.sql.pyspark.worker.logging.enabled\", \"true\")\n",
"spark.createDataFrame([(0,), (5,)], [\"value\"]).select(failing_udf(\"value\")).show()\n",
"\n",
- "logs = spark.table(\"system.session.python_worker_logs\")\n",
+ "logs = spark.tvf.python_worker_logs()\n",
"logs.filter(\"logger = 'error_handler'\").select(\"msg\", \"exception\").show(truncate=False)"
]
},
@@ -1193,7 +1193,7 @@
"df = spark.createDataFrame([(\"hello world\",)], [\"text\"])\n",
"df.lateralJoin(WordSplitter(col(\"text\").outer())).show()\n",
"\n",
- "logs = spark.table(\"system.session.python_worker_logs\")\n",
+ "logs = spark.tvf.python_worker_logs()\n",
"logs.filter(\"logger = 'udtf_logger'\").select(\"msg\", \"context\").show(truncate=False)"
]
},
@@ -1231,7 +1231,7 @@
}
],
"source": [
- "logs = spark.table(\"system.session.python_worker_logs\")\n",
+ "logs = spark.tvf.python_worker_logs()\n",
"\n",
"# Count logs by level\n",
"logs.groupBy(\"level\").count().show()\n",
diff --git a/python/pyspark/sql/connect/tvf.py b/python/pyspark/sql/connect/tvf.py
index 59a4e4fbe344..ac41b0b21725 100644
--- a/python/pyspark/sql/connect/tvf.py
+++ b/python/pyspark/sql/connect/tvf.py
@@ -109,6 +109,11 @@ class TableValuedFunction:
variant_explode_outer.__doc__ = PySparkTableValuedFunction.variant_explode_outer.__doc__
+ def python_worker_logs(self) -> "DataFrame":
+ return self._fn("python_worker_logs")
+
+ python_worker_logs.__doc__ = PySparkTableValuedFunction.python_worker_logs.__doc__
+
def _fn(self, name: str, *args: "Column") -> "DataFrame":
from pyspark.sql.connect.dataframe import DataFrame
from pyspark.sql.connect.plan import UnresolvedTableValuedFunction
diff --git a/python/pyspark/sql/tests/arrow/test_arrow_cogrouped_map.py b/python/pyspark/sql/tests/arrow/test_arrow_cogrouped_map.py
index 6c1c378dcea8..abfd1af7a741 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow_cogrouped_map.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow_cogrouped_map.py
@@ -396,20 +396,20 @@ class CogroupedMapInArrowTestsMixin:
+ [Row(id=2, v1=20, v2=200)],
)
- logs = self.spark.table("system.session.python_worker_logs")
-
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"arrow cogrouped map: {dict(v1=v1, v2=v2)}",
- context={"func_name": func_with_logging.__name__},
- logger="test_arrow_cogrouped_map",
- )
- for v1, v2 in [([10, 30], [100, 300]), ([20], [200])]
- ],
- )
+ logs = self.spark.tvf.python_worker_logs()
+
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"arrow cogrouped map: {dict(v1=v1, v2=v2)}",
+ context={"func_name": func_with_logging.__name__},
+ logger="test_arrow_cogrouped_map",
+ )
+ for v1, v2 in [([10, 30], [100, 300]), ([20], [200])]
+ ],
+ )
class CogroupedMapInArrowTests(CogroupedMapInArrowTestsMixin, ReusedSQLTestCase):
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 829c38385bd0..56a9ed322c15 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py
@@ -416,20 +416,20 @@ class ApplyInArrowTestsMixin:
df,
)
- logs = self.spark.table("system.session.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]]
- ],
- )
+ 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]]
+ ],
+ )
@unittest.skipIf(is_remote_only(), "Requires JVM access")
def test_apply_in_arrow_iter_with_logging(self):
@@ -456,20 +456,20 @@ class ApplyInArrowTestsMixin:
df,
)
- logs = self.spark.table("system.session.python_worker_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]]
+ ],
+ )
class ApplyInArrowTests(ApplyInArrowTestsMixin, ReusedSQLTestCase):
diff --git a/python/pyspark/sql/tests/arrow/test_arrow_map.py b/python/pyspark/sql/tests/arrow/test_arrow_map.py
index 4a56a32fbcdd..31851e8a8adb 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow_map.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow_map.py
@@ -247,12 +247,12 @@ class MapInArrowTestsMixin(object):
[Row(id=i) for i in range(9)],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- self._expected_logs_for_test_map_in_arrow_with_logging(func_with_logging.__name__),
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ self._expected_logs_for_test_map_in_arrow_with_logging(func_with_logging.__name__),
+ )
def _expected_logs_for_test_map_in_arrow_with_logging(self, func_name):
return [
diff --git a/python/pyspark/sql/tests/arrow/test_arrow_udf_grouped_agg.py b/python/pyspark/sql/tests/arrow/test_arrow_udf_grouped_agg.py
index f719b4fb16bd..3d8588ffb7af 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow_udf_grouped_agg.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow_udf_grouped_agg.py
@@ -1044,20 +1044,20 @@ class GroupedAggArrowUDFTestsMixin:
[Row(id=1, result=3.0), Row(id=2, result=18.0)],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"grouped agg arrow udf: {n}",
- context={"func_name": my_grouped_agg_arrow_udf.__name__},
- logger="test_grouped_agg_arrow",
- )
- for n in [2, 3]
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"grouped agg arrow udf: {n}",
+ context={"func_name": my_grouped_agg_arrow_udf.__name__},
+ logger="test_grouped_agg_arrow",
+ )
+ for n in [2, 3]
+ ],
+ )
class GroupedAggArrowUDFTests(GroupedAggArrowUDFTestsMixin, ReusedSQLTestCase):
diff --git a/python/pyspark/sql/tests/arrow/test_arrow_udf_scalar.py b/python/pyspark/sql/tests/arrow/test_arrow_udf_scalar.py
index 05f33a4ae42f..3bd00d2cc921 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow_udf_scalar.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow_udf_scalar.py
@@ -1201,20 +1201,20 @@ class ScalarArrowUDFTestsMixin:
[Row(result=f"scalar_arrow_{i}") for i in range(3)],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"scalar arrow udf: {lst}",
- context={"func_name": my_scalar_arrow_udf.__name__},
- logger="test_scalar_arrow",
- )
- for lst in [[0], [1, 2]]
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"scalar arrow udf: {lst}",
+ context={"func_name": my_scalar_arrow_udf.__name__},
+ logger="test_scalar_arrow",
+ )
+ for lst in [[0], [1, 2]]
+ ],
+ )
@unittest.skipIf(is_remote_only(), "Requires JVM access")
def test_scalar_iter_arrow_udf_with_logging(self):
@@ -1241,20 +1241,20 @@ class ScalarArrowUDFTestsMixin:
[Row(result=f"scalar_iter_arrow_{i}") for i in range(9)],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"scalar iter arrow udf: {lst}",
- context={"func_name": my_scalar_iter_arrow_udf.__name__},
- logger="test_scalar_iter_arrow",
- )
- for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"scalar iter arrow udf: {lst}",
+ context={"func_name": my_scalar_iter_arrow_udf.__name__},
+ logger="test_scalar_iter_arrow",
+ )
+ for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
+ ],
+ )
class ScalarArrowUDFTests(ScalarArrowUDFTestsMixin, ReusedSQLTestCase):
diff --git a/python/pyspark/sql/tests/arrow/test_arrow_udf_window.py b/python/pyspark/sql/tests/arrow/test_arrow_udf_window.py
index 240e34487b00..b0adfbe13186 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow_udf_window.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow_udf_window.py
@@ -834,20 +834,20 @@ class WindowArrowUDFTestsMixin:
],
)
- logs = self.spark.table("system.session.python_worker_logs")
-
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"window arrow udf: {lst}",
- context={"func_name": my_window_arrow_udf.__name__},
- logger="test_window_arrow",
- )
- for lst in [[1.0], [1.0, 2.0], [3.0], [3.0, 5.0], [3.0, 5.0, 10.0]]
- ],
- )
+ logs = self.spark.tvf.python_worker_logs()
+
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"window arrow udf: {lst}",
+ context={"func_name": my_window_arrow_udf.__name__},
+ logger="test_window_arrow",
+ )
+ for lst in [[1.0], [1.0, 2.0], [3.0], [3.0, 5.0], [3.0, 5.0, 10.0]]
+ ],
+ )
class WindowArrowUDFTests(WindowArrowUDFTestsMixin, ReusedSQLTestCase):
diff --git a/python/pyspark/sql/tests/arrow/test_arrow_udtf.py b/python/pyspark/sql/tests/arrow/test_arrow_udtf.py
index 24c5ec0acf91..cc0edda378ab 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow_udtf.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow_udtf.py
@@ -1721,20 +1721,20 @@ class ArrowUDTFTestsMixin:
[Row(id=i, doubled=i * 2) for i in range(9)],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"arrow udtf: {dict(id=lst)}",
- context={"class_name": "TestArrowUDTFWithLogging", "func_name": "eval"},
- logger="test_arrow_udtf",
- )
- for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"arrow udtf: {dict(id=lst)}",
+ context={"class_name": "TestArrowUDTFWithLogging", "func_name": "eval"},
+ logger="test_arrow_udtf",
+ )
+ for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
+ ],
+ )
class ArrowUDTFTests(ArrowUDTFTestsMixin, ReusedSQLTestCase):
diff --git a/python/pyspark/sql/tests/pandas/test_pandas_cogrouped_map.py b/python/pyspark/sql/tests/pandas/test_pandas_cogrouped_map.py
index 8d48bbb794ad..0d91da035497 100644
--- a/python/pyspark/sql/tests/pandas/test_pandas_cogrouped_map.py
+++ b/python/pyspark/sql/tests/pandas/test_pandas_cogrouped_map.py
@@ -742,20 +742,20 @@ class CogroupedApplyInPandasTestsMixin:
+ [Row(id=2, v1=20, v2=200)],
)
- logs = self.spark.table("system.session.python_worker_logs")
-
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"pandas cogrouped map: {dict(v1=v1, v2=v2)}",
- context={"func_name": func_with_logging.__name__},
- logger="test_pandas_cogrouped_map",
- )
- for v1, v2 in [([10, 30], [100, 300]), ([20], [200])]
- ],
- )
+ logs = self.spark.tvf.python_worker_logs()
+
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"pandas cogrouped map: {dict(v1=v1, v2=v2)}",
+ context={"func_name": func_with_logging.__name__},
+ logger="test_pandas_cogrouped_map",
+ )
+ for v1, v2 in [([10, 30], [100, 300]), ([20], [200])]
+ ],
+ )
class CogroupedApplyInPandasTests(CogroupedApplyInPandasTestsMixin, ReusedSQLTestCase):
diff --git a/python/pyspark/sql/tests/pandas/test_pandas_grouped_map.py b/python/pyspark/sql/tests/pandas/test_pandas_grouped_map.py
index 79895bb1a01b..8bf0e7235efc 100644
--- a/python/pyspark/sql/tests/pandas/test_pandas_grouped_map.py
+++ b/python/pyspark/sql/tests/pandas/test_pandas_grouped_map.py
@@ -1000,20 +1000,20 @@ class ApplyInPandasTestsMixin:
df,
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"pandas grouped map: {dict(id=lst, value=[v*10 for v in lst])}",
- context={"func_name": func_with_logging.__name__},
- logger="test_pandas_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"pandas grouped map: {dict(id=lst, value=[v*10 for v in lst])}",
+ context={"func_name": func_with_logging.__name__},
+ logger="test_pandas_grouped_map",
+ )
+ for lst in [[0, 2, 4, 6, 8], [1, 3, 5, 7]]
+ ],
+ )
def test_apply_in_pandas_iterator_basic(self):
df = self.spark.createDataFrame(
diff --git a/python/pyspark/sql/tests/pandas/test_pandas_map.py b/python/pyspark/sql/tests/pandas/test_pandas_map.py
index 5e0e33a05b22..946d56f2fe63 100644
--- a/python/pyspark/sql/tests/pandas/test_pandas_map.py
+++ b/python/pyspark/sql/tests/pandas/test_pandas_map.py
@@ -510,20 +510,20 @@ class MapInPandasTestsMixin:
[Row(id=i) for i in range(9)],
)
- logs = self.spark.table("system.session.python_worker_logs")
-
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"pandas map: {lst}",
- context={"func_name": func_with_logging.__name__},
- logger="test_pandas_map",
- )
- for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
- ],
- )
+ logs = self.spark.tvf.python_worker_logs()
+
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"pandas map: {lst}",
+ context={"func_name": func_with_logging.__name__},
+ logger="test_pandas_map",
+ )
+ for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
+ ],
+ )
class MapInPandasTests(ReusedSQLTestCase, MapInPandasTestsMixin):
diff --git a/python/pyspark/sql/tests/pandas/test_pandas_udf_grouped_agg.py b/python/pyspark/sql/tests/pandas/test_pandas_udf_grouped_agg.py
index 6915e8aee948..4b66dee5b7af 100644
--- a/python/pyspark/sql/tests/pandas/test_pandas_udf_grouped_agg.py
+++ b/python/pyspark/sql/tests/pandas/test_pandas_udf_grouped_agg.py
@@ -846,20 +846,20 @@ class GroupedAggPandasUDFTestsMixin:
[Row(id=1, result=3.0), Row(id=2, result=18.0)],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"grouped agg pandas udf: {n}",
- context={"func_name": my_grouped_agg_pandas_udf.__name__},
- logger="test_grouped_agg_pandas",
- )
- for n in [2, 3]
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"grouped agg pandas udf: {n}",
+ context={"func_name": my_grouped_agg_pandas_udf.__name__},
+ logger="test_grouped_agg_pandas",
+ )
+ for n in [2, 3]
+ ],
+ )
def test_grouped_agg_pandas_udf_with_compression_codec(self):
# Test grouped agg Pandas UDF with different compression codec settings
diff --git a/python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py b/python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py
index 563e0b789a96..b936a9240e52 100644
--- a/python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py
+++ b/python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py
@@ -1935,20 +1935,20 @@ class ScalarPandasUDFTestsMixin:
[Row(result=f"scalar_pandas_{i}") for i in range(3)],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"scalar pandas udf: {lst}",
- context={"func_name": my_scalar_pandas_udf.__name__},
- logger="test_scalar_pandas",
- )
- for lst in [[0], [1, 2]]
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"scalar pandas udf: {lst}",
+ context={"func_name": my_scalar_pandas_udf.__name__},
+ logger="test_scalar_pandas",
+ )
+ for lst in [[0], [1, 2]]
+ ],
+ )
@unittest.skipIf(is_remote_only(), "Requires JVM access")
def test_scalar_iter_pandas_udf_with_logging(self):
@@ -1973,20 +1973,20 @@ class ScalarPandasUDFTestsMixin:
[Row(result=f"scalar_iter_pandas_{i}") for i in range(9)],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"scalar iter pandas udf: {lst}",
- context={"func_name": my_scalar_iter_pandas_udf.__name__},
- logger="test_scalar_iter_pandas",
- )
- for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"scalar iter pandas udf: {lst}",
+ context={"func_name": my_scalar_iter_pandas_udf.__name__},
+ logger="test_scalar_iter_pandas",
+ )
+ for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
+ ],
+ )
def test_scalar_pandas_udf_with_compression_codec(self):
# Test scalar Pandas UDF with different compression codec settings
diff --git a/python/pyspark/sql/tests/pandas/test_pandas_udf_window.py b/python/pyspark/sql/tests/pandas/test_pandas_udf_window.py
index 6fa7e9063836..6e1cbdaf73cf 100644
--- a/python/pyspark/sql/tests/pandas/test_pandas_udf_window.py
+++ b/python/pyspark/sql/tests/pandas/test_pandas_udf_window.py
@@ -664,20 +664,20 @@ class WindowPandasUDFTestsMixin:
],
)
- logs = self.spark.table("system.session.python_worker_logs")
-
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"window pandas udf: {lst}",
- context={"func_name": my_window_pandas_udf.__name__},
- logger="test_window_pandas",
- )
- for lst in [[1.0], [1.0, 2.0], [3.0], [3.0, 5.0], [3.0, 5.0, 10.0]]
- ],
- )
+ logs = self.spark.tvf.python_worker_logs()
+
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"window pandas udf: {lst}",
+ context={"func_name": my_window_pandas_udf.__name__},
+ logger="test_window_pandas",
+ )
+ for lst in [[1.0], [1.0, 2.0], [3.0], [3.0, 5.0], [3.0, 5.0, 10.0]]
+ ],
+ )
class WindowPandasUDFTests(WindowPandasUDFTestsMixin, ReusedSQLTestCase):
diff --git a/python/pyspark/sql/tests/test_python_datasource.py b/python/pyspark/sql/tests/test_python_datasource.py
index 532f68f2c8f2..e5171876656c 100644
--- a/python/pyspark/sql/tests/test_python_datasource.py
+++ b/python/pyspark/sql/tests/test_python_datasource.py
@@ -965,49 +965,49 @@ class BasePythonDataSourceTestsMixin:
],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=msg,
- context=context,
- logger="test_data_source_reader",
- )
- for msg, context in [
- (
- "TestJsonDataSource.__init__: ['path']",
- {"class_name": "TestJsonDataSource", "func_name": "__init__"},
- ),
- (
- "TestJsonDataSource.name",
- {"class_name": "TestJsonDataSource", "func_name": "name"},
- ),
- (
- "TestJsonDataSource.schema",
- {"class_name": "TestJsonDataSource", "func_name": "schema"},
- ),
- (
- "TestJsonDataSource.reader: ['name', 'age']",
- {"class_name": "TestJsonDataSource", "func_name": "reader"},
- ),
- (
- "TestJsonReader.__init__: ['path']",
- {"class_name": "TestJsonDataSource", "func_name": "reader"},
- ),
- (
- "TestJsonReader.partitions",
- {"class_name": "TestJsonReader", "func_name": "partitions"},
- ),
- (
- "TestJsonReader.read: None",
- {"class_name": "TestJsonReader", "func_name": "read"},
- ),
- ]
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=msg,
+ context=context,
+ logger="test_data_source_reader",
+ )
+ for msg, context in [
+ (
+ "TestJsonDataSource.__init__: ['path']",
+ {"class_name": "TestJsonDataSource", "func_name": "__init__"},
+ ),
+ (
+ "TestJsonDataSource.name",
+ {"class_name": "TestJsonDataSource", "func_name": "name"},
+ ),
+ (
+ "TestJsonDataSource.schema",
+ {"class_name": "TestJsonDataSource", "func_name": "schema"},
+ ),
+ (
+ "TestJsonDataSource.reader: ['name', 'age']",
+ {"class_name": "TestJsonDataSource", "func_name": "reader"},
+ ),
+ (
+ "TestJsonReader.__init__: ['path']",
+ {"class_name": "TestJsonDataSource", "func_name": "reader"},
+ ),
+ (
+ "TestJsonReader.partitions",
+ {"class_name": "TestJsonReader", "func_name": "partitions"},
+ ),
+ (
+ "TestJsonReader.read: None",
+ {"class_name": "TestJsonReader", "func_name": "read"},
+ ),
+ ]
+ ],
+ )
@unittest.skipIf(is_remote_only(), "Requires JVM access")
def test_data_source_reader_pushdown_with_logging(self):
@@ -1072,53 +1072,53 @@ class BasePythonDataSourceTestsMixin:
],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=msg,
- context=context,
- logger="test_data_source_reader_pushdown",
- )
- for msg, context in [
- (
- "TestJsonDataSource.__init__: ['path']",
- {"class_name": "TestJsonDataSource", "func_name": "__init__"},
- ),
- (
- "TestJsonDataSource.name",
- {"class_name": "TestJsonDataSource", "func_name": "name"},
- ),
- (
- "TestJsonDataSource.schema",
- {"class_name": "TestJsonDataSource", "func_name": "schema"},
- ),
- (
- "TestJsonDataSource.reader: ['name', 'age']",
- {"class_name": "TestJsonDataSource", "func_name": "reader"},
- ),
- (
- "TestJsonReader.pushFilters: [IsNotNull(attribute=('age',))]",
- {"class_name": "TestJsonReader", "func_name": "pushFilters"},
- ),
- (
- "TestJsonReader.__init__: ['path']",
- {"class_name": "TestJsonDataSource", "func_name": "reader"},
- ),
- (
- "TestJsonReader.partitions",
- {"class_name": "TestJsonReader", "func_name": "partitions"},
- ),
- (
- "TestJsonReader.read: None",
- {"class_name": "TestJsonReader", "func_name": "read"},
- ),
- ]
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=msg,
+ context=context,
+ logger="test_data_source_reader_pushdown",
+ )
+ for msg, context in [
+ (
+ "TestJsonDataSource.__init__: ['path']",
+ {"class_name": "TestJsonDataSource", "func_name": "__init__"},
+ ),
+ (
+ "TestJsonDataSource.name",
+ {"class_name": "TestJsonDataSource", "func_name": "name"},
+ ),
+ (
+ "TestJsonDataSource.schema",
+ {"class_name": "TestJsonDataSource", "func_name": "schema"},
+ ),
+ (
+ "TestJsonDataSource.reader: ['name', 'age']",
+ {"class_name": "TestJsonDataSource", "func_name": "reader"},
+ ),
+ (
+ "TestJsonReader.pushFilters: [IsNotNull(attribute=('age',))]",
+ {"class_name": "TestJsonReader", "func_name": "pushFilters"},
+ ),
+ (
+ "TestJsonReader.__init__: ['path']",
+ {"class_name": "TestJsonDataSource", "func_name": "reader"},
+ ),
+ (
+ "TestJsonReader.partitions",
+ {"class_name": "TestJsonReader", "func_name": "partitions"},
+ ),
+ (
+ "TestJsonReader.read: None",
+ {"class_name": "TestJsonReader", "func_name": "read"},
+ ),
+ ]
+ ],
+ )
@unittest.skipIf(is_remote_only(), "Requires JVM access")
def test_data_source_writer_with_logging(self):
@@ -1197,69 +1197,69 @@ class BasePythonDataSourceTestsMixin:
with self.assertRaises(Exception, msg="abort test"):
df.write.format("my-json").mode("append").option("abort", "true").save(d)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=msg,
- context=context,
- logger="test_datasource_writer",
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=msg,
+ context=context,
+ logger="test_datasource_writer",
+ )
+ for msg, context in [
+ (
+ "TestJsonDataSource.name",
+ {"class_name": "TestJsonDataSource", "func_name": "name"},
+ ),
+ (
+ "TestJsonDataSource.writer: (['name', 'age'], {True})",
+ {"class_name": "TestJsonDataSource", "func_name": "writer"},
+ ),
+ (
+ "TestJsonWriter.__init__: ['path']",
+ {"class_name": "TestJsonDataSource", "func_name": "writer"},
+ ),
+ (
+ "TestJsonWriter.write: 1, [{'name': 'Diana', 'age': 28}]",
+ {"class_name": "TestJsonWriter", "func_name": "write"},
+ ),
+ (
+ "TestJsonWriter.write: 1, [{'name': 'Charlie', 'age': 35}]",
+ {"class_name": "TestJsonWriter", "func_name": "write"},
+ ),
+ (
+ "TestJsonWriter.commit: 2",
+ {"class_name": "TestJsonWriter", "func_name": "commit"},
+ ),
+ (
+ "TestJsonDataSource.name",
+ {"class_name": "TestJsonDataSource", "func_name": "name"},
+ ),
+ (
+ "TestJsonDataSource.writer: (['name', 'age'], {False})",
+ {"class_name": "TestJsonDataSource", "func_name": "writer"},
+ ),
+ (
+ "TestJsonWriter.__init__: ['abort', 'path']",
+ {"class_name": "TestJsonDataSource", "func_name": "writer"},
+ ),
+ (
+ "TestJsonWriter.write: abort test",
+ {"class_name": "TestJsonWriter", "func_name": "write"},
+ ),
+ (
+ "TestJsonWriter.write: abort test",
+ {"class_name": "TestJsonWriter", "func_name": "write"},
+ ),
+ (
+ "TestJsonWriter.abort",
+ {"class_name": "TestJsonWriter", "func_name": "abort"},
+ ),
+ ]
+ ],
)
- for msg, context in [
- (
- "TestJsonDataSource.name",
- {"class_name": "TestJsonDataSource", "func_name": "name"},
- ),
- (
- "TestJsonDataSource.writer: (['name', 'age'], {True})",
- {"class_name": "TestJsonDataSource", "func_name": "writer"},
- ),
- (
- "TestJsonWriter.__init__: ['path']",
- {"class_name": "TestJsonDataSource", "func_name": "writer"},
- ),
- (
- "TestJsonWriter.write: 1, [{'name': 'Diana', 'age': 28}]",
- {"class_name": "TestJsonWriter", "func_name": "write"},
- ),
- (
- "TestJsonWriter.write: 1, [{'name': 'Charlie', 'age': 35}]",
- {"class_name": "TestJsonWriter", "func_name": "write"},
- ),
- (
- "TestJsonWriter.commit: 2",
- {"class_name": "TestJsonWriter", "func_name": "commit"},
- ),
- (
- "TestJsonDataSource.name",
- {"class_name": "TestJsonDataSource", "func_name": "name"},
- ),
- (
- "TestJsonDataSource.writer: (['name', 'age'], {False})",
- {"class_name": "TestJsonDataSource", "func_name": "writer"},
- ),
- (
- "TestJsonWriter.__init__: ['abort', 'path']",
- {"class_name": "TestJsonDataSource", "func_name": "writer"},
- ),
- (
- "TestJsonWriter.write: abort test",
- {"class_name": "TestJsonWriter", "func_name": "write"},
- ),
- (
- "TestJsonWriter.write: abort test",
- {"class_name": "TestJsonWriter", "func_name": "write"},
- ),
- (
- "TestJsonWriter.abort",
- {"class_name": "TestJsonWriter", "func_name": "abort"},
- ),
- ]
- ],
- )
class PythonDataSourceTests(BasePythonDataSourceTestsMixin, ReusedSQLTestCase):
diff --git a/python/pyspark/sql/tests/test_udf.py b/python/pyspark/sql/tests/test_udf.py
index 0ceb745c5860..9c5fa2ad1bba 100644
--- a/python/pyspark/sql/tests/test_udf.py
+++ b/python/pyspark/sql/tests/test_udf.py
@@ -1573,50 +1573,66 @@ class BaseUDFTestsMixin(object):
logger.exception("exception")
return "x"
+ # The TVF is not available when the feature is disabled.
+ with self.assertRaises(AnalysisException) as pe:
+ self.spark.tvf.python_worker_logs().count()
+
+ self.check_error(
+ exception=pe.exception,
+ errorClass="FEATURE_NOT_ENABLED",
+ messageParameters={
+ "featureName": "Python Worker Logging",
+ "configKey": "spark.sql.pyspark.worker.logging.enabled",
+ "configValue": "true",
+ },
+ )
+
# Logging is disabled by default
assertDataFrameEqual(
self.spark.range(1).select(my_udf().alias("result")), [Row(result="x")]
)
- self.assertEqual(self.spark.table("system.session.python_worker_logs").count(), 0)
with self.sql_conf({"spark.sql.pyspark.worker.logging.enabled": "true"}):
+ # The logs were not collected when the feature was disabled.
+ self.assertEqual(self.spark.tvf.python_worker_logs().count(), 0)
+
assertDataFrameEqual(
self.spark.range(1).select(my_udf().alias("result")), [Row(result="x")]
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="INFO",
- msg="print to stdout ❤",
- context={"func_name": my_udf.__name__},
- logger="stdout",
- ),
- Row(
- level="ERROR",
- msg="print to stderr 😀",
- context={"func_name": my_udf.__name__},
- logger="stderr",
- ),
- Row(
- level="WARNING",
- msg="custom context",
- context={"func_name": my_udf.__name__, "abc": "123"},
- logger="test",
- ),
- Row(
- level="ERROR",
- msg="exception",
- context={"func_name": my_udf.__name__},
- logger="test",
- ),
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="INFO",
+ msg="print to stdout ❤",
+ context={"func_name": my_udf.__name__},
+ logger="stdout",
+ ),
+ Row(
+ level="ERROR",
+ msg="print to stderr 😀",
+ context={"func_name": my_udf.__name__},
+ logger="stderr",
+ ),
+ Row(
+ level="WARNING",
+ msg="custom context",
+ context={"func_name": my_udf.__name__, "abc": "123"},
+ logger="test",
+ ),
+ Row(
+ level="ERROR",
+ msg="exception",
+ context={"func_name": my_udf.__name__},
+ logger="test",
+ ),
+ ],
+ )
- self.assertEqual(logs.where("exception is not null").select("exception").count(), 1)
+ self.assertEqual(logs.where("exception is not null").select("exception").count(), 1)
@unittest.skipIf(is_remote_only(), "Requires JVM access")
def test_multiple_udfs_with_logging(self):
@@ -1638,25 +1654,25 @@ class BaseUDFTestsMixin(object):
[Row(result="x", result2="y")],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg="test1",
- context={"func_name": my_udf1.__name__},
- logger="test1",
- ),
- Row(
- level="WARNING",
- msg="test2",
- context={"func_name": my_udf2.__name__},
- logger="test2",
- ),
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg="test1",
+ context={"func_name": my_udf1.__name__},
+ logger="test1",
+ ),
+ Row(
+ level="WARNING",
+ msg="test2",
+ context={"func_name": my_udf2.__name__},
+ logger="test2",
+ ),
+ ],
+ )
@unittest.skipIf(is_remote_only(), "Requires JVM access")
def test_udf_with_pyspark_logger(self):
@@ -1672,20 +1688,20 @@ class BaseUDFTestsMixin(object):
[Row(result=str(i)) for i in range(2)],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg="PySparkLogger test",
- context={"func_name": my_udf.__name__, "x": str(i)},
- logger="PySparkLogger",
- )
- for i in range(2)
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg="PySparkLogger test",
+ context={"func_name": my_udf.__name__, "x": str(i)},
+ logger="PySparkLogger",
+ )
+ for i in range(2)
+ ],
+ )
class UDFTests(BaseUDFTestsMixin, ReusedSQLTestCase):
diff --git a/python/pyspark/sql/tests/test_udtf.py b/python/pyspark/sql/tests/test_udtf.py
index 70623a5f62d4..5ded5aa67b4e 100644
--- a/python/pyspark/sql/tests/test_udtf.py
+++ b/python/pyspark/sql/tests/test_udtf.py
@@ -3079,20 +3079,20 @@ class BaseUDTFTestsMixin:
[Row(x=x, a=x * 2, b=x + 10) for x in [5, 10]],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"udtf with logging: {x}",
- context={"class_name": "TestUDTFWithLogging", "func_name": "eval"},
- logger="test_udtf",
- )
- for x in [5, 10]
- ],
- )
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"udtf with logging: {x}",
+ context={"class_name": "TestUDTFWithLogging", "func_name": "eval"},
+ logger="test_udtf",
+ )
+ for x in [5, 10]
+ ],
+ )
@unittest.skipIf(is_remote_only(), "Requires JVM access")
def test_udtf_analyze_with_logging(self):
@@ -3115,26 +3115,26 @@ class BaseUDTFTestsMixin:
[Row(x=x, a=x * 2, b=x + 10) for x in [5, 10]],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select(
- "level",
- "msg",
- col("context.class_name").alias("context_class_name"),
- col("context.func_name").alias("context_func_name"),
- "logger",
- ).distinct(),
- [
- Row(
- level="WARNING",
- msg='udtf analyze: "long"',
- context_class_name="TestUDTFWithLogging",
- context_func_name="analyze",
- logger="test_udtf",
- )
- ],
- )
+ assertDataFrameEqual(
+ logs.select(
+ "level",
+ "msg",
+ col("context.class_name").alias("context_class_name"),
+ col("context.func_name").alias("context_func_name"),
+ "logger",
+ ).distinct(),
+ [
+ Row(
+ level="WARNING",
+ msg='udtf analyze: "long"',
+ context_class_name="TestUDTFWithLogging",
+ context_func_name="analyze",
+ logger="test_udtf",
+ )
+ ],
+ )
@unittest.skipIf(is_remote_only(), "Requires JVM access")
def test_udtf_analyze_with_pyspark_logger(self):
@@ -3157,28 +3157,28 @@ class BaseUDTFTestsMixin:
[Row(x=x, a=x * 2, b=x + 10) for x in [5, 10]],
)
- logs = self.spark.table("system.session.python_worker_logs")
+ logs = self.spark.tvf.python_worker_logs()
- assertDataFrameEqual(
- logs.select(
- "level",
- "msg",
- col("context.class_name").alias("context_class_name"),
- col("context.func_name").alias("context_func_name"),
- col("context.dt").alias("context_dt"),
- "logger",
- ).distinct(),
- [
- Row(
- level="WARNING",
- msg='udtf analyze: "long"',
- context_class_name="TestUDTFWithLogging",
- context_func_name="analyze",
- context_dt='"long"',
- logger="PySparkLogger",
- )
- ],
- )
+ assertDataFrameEqual(
+ logs.select(
+ "level",
+ "msg",
+ col("context.class_name").alias("context_class_name"),
+ col("context.func_name").alias("context_func_name"),
+ col("context.dt").alias("context_dt"),
+ "logger",
+ ).distinct(),
+ [
+ Row(
+ level="WARNING",
+ msg='udtf analyze: "long"',
+ context_class_name="TestUDTFWithLogging",
+ context_func_name="analyze",
+ context_dt='"long"',
+ logger="PySparkLogger",
+ )
+ ],
+ )
class UDTFTests(BaseUDTFTestsMixin, ReusedSQLTestCase):
diff --git a/python/pyspark/sql/tvf.py b/python/pyspark/sql/tvf.py
index b34877b03311..c21e3751655c 100644
--- a/python/pyspark/sql/tvf.py
+++ b/python/pyspark/sql/tvf.py
@@ -677,6 +677,45 @@ class TableValuedFunction:
"""
return self._fn("variant_explode_outer", input)
+ def python_worker_logs(self) -> DataFrame:
+ """
+ Returns a DataFrame of logs collected from Python workers.
+
+ .. versionadded:: 4.1.0
+
+ Returns
+ -------
+ :class:`DataFrame`
+
+ Examples
+ --------
+ >>> import pyspark.sql.functions as sf
+ >>> import logging
+ >>>
+ >>> @sf.udf("string")
+ ... def my_udf(x):
+ ... logger = logging.getLogger("my_custom_logger")
+ ... logger.warning("This is a warning")
+ ... return str(x)
+ ...
+ >>> spark.conf.set("spark.sql.pyspark.worker.logging.enabled", "true")
+ >>> spark.range(1).select(my_udf("id")).show()
+ +----------+
+ |my_udf(id)|
+ +----------+
+ | 0|
+ +----------+
+ >>> spark.tvf.python_worker_logs().select(
+ ... "level", "msg", "context", "logger"
+ ... ).show(truncate=False) # doctest: +SKIP
+ +-------+-----------------+---------------------+----------------+
+ |level |msg |context |logger |
+ +-------+-----------------+---------------------+----------------+
+ |WARNING|This is a warning|{func_name -> my_udf}|my_custom_logger|
+ +-------+-----------------+---------------------+----------------+
+ """
+ return self._fn("python_worker_logs")
+
def _fn(self, functionName: str, *args: Column) -> DataFrame:
from pyspark.sql.classic.column import _to_java_column
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/TableValuedFunction.scala b/sql/api/src/main/scala/org/apache/spark/sql/TableValuedFunction.scala
index d2479ce9879c..56f2f75c2c19 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/TableValuedFunction.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/TableValuedFunction.scala
@@ -181,4 +181,12 @@ abstract class TableValuedFunction {
* @since 4.0.0
*/
def variant_explode_outer(input: Column): Dataset[Row]
+
+ /**
+ * Returns a `DataFrame` of logs collected from Python workers.
+ *
+ * @group table_funcs
+ * @since 4.1.0
+ */
+ def python_worker_logs(): Dataset[Row]
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
index e6e713fab7d9..36e40306be7d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
@@ -33,7 +33,7 @@ import org.apache.spark.sql.catalyst.expressions.aggregate._
import org.apache.spark.sql.catalyst.expressions.st._
import org.apache.spark.sql.catalyst.expressions.variant._
import org.apache.spark.sql.catalyst.expressions.xml._
-import org.apache.spark.sql.catalyst.plans.logical.{FunctionBuilderBase, Generate, LogicalPlan, OneRowRelation, Range}
+import org.apache.spark.sql.catalyst.plans.logical.{FunctionBuilderBase, Generate, LogicalPlan, OneRowRelation, PythonWorkerLogs, Range}
import org.apache.spark.sql.catalyst.trees.TreeNodeTag
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.SQLConf
@@ -1233,7 +1233,8 @@ object TableFunctionRegistry {
generator[Collations]("collations"),
generator[SQLKeywords]("sql_keywords"),
generatorBuilder("variant_explode", VariantExplodeGeneratorBuilder),
- generatorBuilder("variant_explode_outer", VariantExplodeOuterGeneratorBuilder)
+ generatorBuilder("variant_explode_outer", VariantExplodeOuterGeneratorBuilder),
+ PythonWorkerLogs.functionBuilder
)
val builtin: SimpleTableFunctionRegistry = {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
index ea5836b8ec2d..6ff40da88ed1 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
@@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.catalog.{
TemporaryViewRelation,
UnresolvedCatalogRelation
}
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, PythonWorkerLogs, SubqueryAlias}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
import org.apache.spark.sql.connector.catalog.{
CatalogManager,
@@ -112,8 +112,6 @@ class RelationResolution(override val catalogManager: CatalogManager)
u.isStreaming,
finalTimeTravelSpec.isDefined
).orElse {
- resolveSystemSessionView(u.multipartIdentifier)
- }.orElse {
expandIdentifier(u.multipartIdentifier) match {
case CatalogAndIdentifier(catalog, ident) =>
val key = toCacheKey(catalog, ident, finalTimeTravelSpec)
@@ -238,22 +236,6 @@ class RelationResolution(override val catalogManager: CatalogManager)
}
}
- private def isSystemSessionIdentifier(identifier: Seq[String]): Boolean = {
- identifier.length > 2 &&
- identifier(0).equalsIgnoreCase(CatalogManager.SYSTEM_CATALOG_NAME) &&
- identifier(1).equalsIgnoreCase(CatalogManager.SESSION_NAMESPACE)
- }
-
- private def resolveSystemSessionView(
- identifier: Seq[String]): Option[LogicalPlan] = {
- if (isSystemSessionIdentifier(identifier)) {
- Option(identifier.drop(2)).collect {
- case Seq(viewName) if viewName.equalsIgnoreCase(PythonWorkerLogs.ViewName) =>
- PythonWorkerLogs.viewDefinition()
- }
- } else None
- }
-
private def toCacheKey(
catalog: CatalogPlugin,
ident: Identifier,
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/pythonLogicalOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/pythonLogicalOperators.scala
index 802678bc2c6c..bcfcae2ee16c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/pythonLogicalOperators.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/pythonLogicalOperators.scala
@@ -19,11 +19,13 @@ package org.apache.spark.sql.catalyst.plans.logical
import org.apache.spark.resource.ResourceProfile
import org.apache.spark.sql.catalyst.SQLConfHelper
-import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, UnresolvedAttribute, UnresolvedStar}
-import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, Expression, JsonToStructs, PythonUDF, PythonUDTF}
+import org.apache.spark.sql.catalyst.analysis.{FunctionRegistryBase, MultiInstanceRelation, UnresolvedAttribute, UnresolvedStar}
+import org.apache.spark.sql.catalyst.analysis.TableFunctionRegistry.TableFunctionBuilder
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, Expression, ExpressionDescription, ExpressionInfo, JsonToStructs, PythonUDF, PythonUDTF}
import org.apache.spark.sql.catalyst.trees.TreePattern._
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.truncatedString
+import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.streaming.{GroupStateTimeout, OutputMode, TimeMode}
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.util.LogUtils
@@ -386,9 +388,26 @@ case class AttachDistributedSequence(
}
}
+// scalastyle:off line.contains.tab line.size.limit
+@ExpressionDescription(
+ usage = """
+ _FUNC_() - Returns a table of logs collected from Python workers.
+ """,
+ examples = """
+ Examples:
+ > SET spark.sql.pyspark.worker.logging.enabled=true;
+ spark.sql.pyspark.worker.logging.enabled true
+ > SELECT * FROM _FUNC_();
+
+ """,
+ since = "4.1.0",
+ group = "table_funcs")
+// scalastyle:on line.contains.tab line.size.limit
case class PythonWorkerLogs(jsonAttr: Attribute)
extends LeafNode with MultiInstanceRelation with SQLConfHelper {
+ def this() = this(DataTypeUtils.toAttribute(StructField("message", StringType)))
+
override def output: Seq[Attribute] = Seq(jsonAttr)
override def newInstance(): PythonWorkerLogs =
@@ -403,23 +422,28 @@ case class PythonWorkerLogs(jsonAttr: Attribute)
)
}
-object PythonWorkerLogs {
- val ViewName = "python_worker_logs"
-
- def apply(): LogicalPlan = {
- PythonWorkerLogs(DataTypeUtils.toAttribute(StructField("message", StringType)))
- }
-
- def viewDefinition(): LogicalPlan = {
- Project(
- Seq(UnresolvedStar(Some(Seq("from_json")))),
- Project(
- Seq(Alias(
- JsonToStructs(
- schema = StructType.fromDDL(LogUtils.SPARK_LOG_SCHEMA),
- options = Map.empty,
- child = UnresolvedAttribute("message")),
- "from_json")()),
- PythonWorkerLogs()))
+object PythonWorkerLogs extends SQLConfHelper {
+ val TableFunctionName = "python_worker_logs"
+
+ val functionBuilder: (String, (ExpressionInfo, TableFunctionBuilder)) = {
+ val (info, builder) = FunctionRegistryBase.build[PythonWorkerLogs](
+ TableFunctionName, None)
+ val funcBuilder = (expressions: Seq[Expression]) => {
+ if (conf.pythonWorkerLoggingEnabled) {
+ Project(
+ Seq(UnresolvedStar(Some(Seq("from_json")))),
+ Project(
+ Seq(Alias(
+ JsonToStructs(
+ schema = StructType.fromDDL(LogUtils.SPARK_LOG_SCHEMA),
+ options = Map.empty,
+ child = UnresolvedAttribute("message")),
+ "from_json")()),
+ builder(expressions)))
+ } else {
+ throw QueryCompilationErrors.pythonWorkerLoggingNotEnabledError()
+ }
+ }
+ TableFunctionName -> (info, funcBuilder)
}
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index 246973bbd246..d00d639eb5de 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -4419,4 +4419,15 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
origin = origin
)
}
+
+ def pythonWorkerLoggingNotEnabledError(): Throwable = {
+ new AnalysisException(
+ errorClass = "FEATURE_NOT_ENABLED",
+ messageParameters = Map(
+ "featureName" -> "Python Worker Logging",
+ "configKey" -> "spark.sql.pyspark.worker.logging.enabled",
+ "configValue" -> "true"
+ )
+ )
+ }
}
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/TableValuedFunction.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/TableValuedFunction.scala
index 05fc4b441f98..e773f66cd6d0 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/TableValuedFunction.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/TableValuedFunction.scala
@@ -102,4 +102,8 @@ class TableValuedFunction(sparkSession: SparkSession) extends sql.TableValuedFun
/** @inheritdoc */
override def variant_explode_outer(input: Column): Dataset[Row] =
fn("variant_explode_outer", Seq(input))
+
+ /** @inheritdoc */
+ override def python_worker_logs(): Dataset[Row] =
+ fn("python_worker_logs", Seq.empty)
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/TableValuedFunction.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/TableValuedFunction.scala
index d2034033fee7..839faa85aed4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/TableValuedFunction.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/TableValuedFunction.scala
@@ -97,4 +97,8 @@ class TableValuedFunction(sparkSession: SparkSession)
/** @inheritdoc */
override def variant_explode_outer(input: Column): Dataset[Row] =
fn("variant_explode_outer", Seq(input))
+
+ /** @inheritdoc */
+ override def python_worker_logs(): Dataset[Row] =
+ fn("python_worker_logs", Seq.empty)
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonWorkerLogsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonWorkerLogsSuite.scala
index 2f363d3e8954..ed195afc7e4d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonWorkerLogsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonWorkerLogsSuite.scala
@@ -19,7 +19,9 @@ package org.apache.spark.sql.execution.python
import java.util.UUID
-import org.apache.spark.sql.{QueryTest, Row}
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.StructType
import org.apache.spark.storage.{PythonWorkerLogBlockId, PythonWorkerLogBlockIdGenerator, PythonWorkerLogLine}
@@ -28,6 +30,9 @@ import org.apache.spark.util.LogUtils
class PythonWorkerLogsSuite extends QueryTest with SharedSparkSession {
import testImplicits._
+ override def sparkConf: SparkConf =
+ super.sparkConf.set(SQLConf.PYTHON_WORKER_LOGGING_ENABLED.key, "true")
+
protected override def afterEach(): Unit = {
try {
val blockManager = spark.sparkContext.env.blockManager
@@ -38,8 +43,25 @@ class PythonWorkerLogsSuite extends QueryTest with SharedSparkSession {
}
}
+ test("fail if logging is not enabled") {
+ withSQLConf(SQLConf.PYTHON_WORKER_LOGGING_ENABLED.key -> "false") {
+ val ex = intercept[AnalysisException] {
+ spark.tvf.python_worker_logs().collect()
+ }
+ checkError(
+ exception = ex,
+ condition = "FEATURE_NOT_ENABLED",
+ parameters = Map(
+ "featureName" -> "Python Worker Logging",
+ "configKey" -> "spark.sql.pyspark.worker.logging.enabled",
+ "configValue" -> "true"
+ )
+ )
+ }
+ }
+
test("schema") {
- val schema = spark.table("system.session.python_worker_logs").schema
+ val schema = spark.tvf.python_worker_logs().schema
assert(schema == StructType.fromDDL(LogUtils.SPARK_LOG_SCHEMA))
}
@@ -57,7 +79,7 @@ class PythonWorkerLogsSuite extends QueryTest with SharedSparkSession {
test("read logs") {
prepareLogs(spark.sessionUUID)
- val df = spark.table("system.session.python_worker_logs")
+ val df = spark.tvf.python_worker_logs()
assert(df.count() == 2)
checkAnswer(
df.select($"level", $"msg"),
@@ -67,7 +89,7 @@ class PythonWorkerLogsSuite extends QueryTest with SharedSparkSession {
test("can't read logs for another session") {
prepareLogs(UUID.randomUUID.toString)
- val df = spark.table("system.session.python_worker_logs")
+ val df = spark.tvf.python_worker_logs()
assert(df.count() == 0)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org