You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2023/01/31 00:06:58 UTC
[spark] branch branch-3.4 updated: [SPARK-42192][PYTHON] Migrate the `TypeError` from `pyspark/sql/dataframe.py` into `PySparkTypeError`
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.4 by this push:
new 42e183f3859 [SPARK-42192][PYTHON] Migrate the `TypeError` from `pyspark/sql/dataframe.py` into `PySparkTypeError`
42e183f3859 is described below
commit 42e183f3859600c39a27940377d9ccb7942184bd
Author: itholic <ha...@databricks.com>
AuthorDate: Tue Jan 31 09:06:27 2023 +0900
[SPARK-42192][PYTHON] Migrate the `TypeError` from `pyspark/sql/dataframe.py` into `PySparkTypeError`
### What changes were proposed in this pull request?
This PR proposes to migrate the `TypeError` from `pyspark/sql/dataframe.py` into `PySparkTypeError`.
### Why are the changes needed?
To integrate the errors generated by PySpark into single path.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added & updated UTs.
Closes #39785 from itholic/error_dataframe.
Authored-by: itholic <ha...@databricks.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
(cherry picked from commit a1362b19094fc8e8da3409feb0e20576c0515c7f)
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
python/pyspark/errors/error_classes.py | 77 +++++-
python/pyspark/sql/connect/dataframe.py | 111 +++++++--
python/pyspark/sql/dataframe.py | 263 ++++++++++++++++-----
.../sql/tests/connect/test_connect_basic.py | 14 +-
python/pyspark/sql/tests/test_dataframe.py | 143 +++++++++--
python/pyspark/sql/tests/test_functions.py | 50 ++++
6 files changed, 561 insertions(+), 97 deletions(-)
diff --git a/python/pyspark/errors/error_classes.py b/python/pyspark/errors/error_classes.py
index dd86f06505b..b30b673974c 100644
--- a/python/pyspark/errors/error_classes.py
+++ b/python/pyspark/errors/error_classes.py
@@ -19,14 +19,34 @@ import json
ERROR_CLASSES_JSON = """
{
+ "ARGUMENT_REQUIRED": {
+ "message": [
+ "Argument `<arg_name>` is required when <condition>."
+ ]
+ },
"COLUMN_IN_LIST": {
"message": [
"<func_name> does not allow a Column in a list."
]
},
+ "DISALLOWED_TYPE_FOR_CONTAINER" : {
+ "message" : [
+ "Argument `<arg_name>`(type: <arg_type>) should only contain a type in [<allowed_types>], got <return_type>"
+ ]
+ },
"HIGHER_ORDER_FUNCTION_SHOULD_RETURN_COLUMN" : {
"message" : [
- "Function `<func_name>` should return Column, got <return_type>"
+ "Function `<func_name>` should return Column, got <return_type>."
+ ]
+ },
+ "NOT_AN_INTEGER" : {
+ "message" : [
+ "Argument `<arg_name>` should be a integer, got <arg_type>."
+ ]
+ },
+ "NOT_A_BOOLEAN" : {
+ "message" : [
+ "Argument `<arg_name>` should be a boolean, got <arg_type>."
]
},
"NOT_A_COLUMN" : {
@@ -39,16 +59,51 @@ ERROR_CLASSES_JSON = """
"Argument `<arg_name>` must be a DataFrame, got <arg_type>."
]
},
+ "NOT_A_DATAFRAME" : {
+ "message" : [
+ "Argument `<arg_name>` should be a DataFrame, got <arg_type>."
+ ]
+ },
+ "NOT_A_DICT" : {
+ "message" : [
+ "Argument `<arg_name>` should be a dict, got <arg_type>."
+ ]
+ },
"NOT_A_STRING" : {
"message" : [
"Argument `<arg_name>` should be a str, got <arg_type>."
]
},
+ "NOT_BOOL_OR_DICT_OR_FLOAT_OR_INTEGER_OR_LIST_OR_STRING_OR_TUPLE" : {
+ "message" : [
+ "Argument `<arg_name>` should be a boolean, dict, float, integer, string or tuple, got <arg_type>."
+ ]
+ },
+ "NOT_BOOL_OR_DICT_OR_FLOAT_OR_INTEGER_OR_STRING" : {
+ "message" : [
+ "Argument `<arg_name>` should be a boolean, dict, float, integer or string, got <arg_type>."
+ ]
+ },
+ "NOT_BOOL_OR_LIST" : {
+ "message" : [
+ "Argument `<arg_name>` should be a boolean or list, got <arg_type>."
+ ]
+ },
+ "NOT_BOOL_OR_STRING" : {
+ "message" : [
+ "Argument `<arg_name>` should be a boolean or string, got <arg_type>."
+ ]
+ },
"NOT_COLUMN_OR_DATATYPE_OR_STRING" : {
"message" : [
"Argument `<arg_name>` should be a Column or str or DataType, but got <arg_type>."
]
},
+ "NOT_COLUMN_OR_FLOAT_OR_INTEGER_OR_LIST_OR_STRING" : {
+ "message" : [
+ "Argument `<arg_name>` should be a column, float, integer, list or string, got <arg_type>."
+ ]
+ },
"NOT_COLUMN_OR_INTEGER" : {
"message" : [
"Argument `<arg_name>` should be a Column or int, got <arg_type>."
@@ -64,6 +119,26 @@ ERROR_CLASSES_JSON = """
"Argument `<arg_name>` should be a Column or str, got <arg_type>."
]
},
+ "NOT_FLOAT_OR_INTEGER" : {
+ "message" : [
+ "Argument `<arg_name>` should be a float or integer, got <arg_type>."
+ ]
+ },
+ "NOT_FLOAT_OR_INTEGER_OR_LIST_OR_STRING" : {
+ "message" : [
+ "Argument `<arg_name>` should be a float, integer, list or string, got <arg_type>."
+ ]
+ },
+ "NOT_LIST_OR_STRING_OR_TUPLE" : {
+ "message" : [
+ "Argument `<arg_name>` should be a list, string or tuple, got <arg_type>."
+ ]
+ },
+ "NOT_LIST_OR_TUPLE" : {
+ "message" : [
+ "Argument `<arg_name>` should be a list or tuple, got <arg_type>."
+ ]
+ },
"UNSUPPORTED_NUMPY_ARRAY_SCALAR" : {
"message" : [
"The type of array scalar '<dtype>' is not supported."
diff --git a/python/pyspark/sql/connect/dataframe.py b/python/pyspark/sql/connect/dataframe.py
index d82862a870b..482c48f0c91 100644
--- a/python/pyspark/sql/connect/dataframe.py
+++ b/python/pyspark/sql/connect/dataframe.py
@@ -46,6 +46,7 @@ from pyspark.sql.dataframe import (
DataFrameStatFunctions as PySparkDataFrameStatFunctions,
)
+from pyspark.errors import PySparkTypeError
import pyspark.sql.connect.plan as plan
from pyspark.sql.connect.group import GroupedData
from pyspark.sql.connect.readwriter import DataFrameWriter
@@ -140,6 +141,11 @@ class DataFrame:
alias.__doc__ = PySparkDataFrame.alias.__doc__
def colRegex(self, colName: str) -> Column:
+ if not isinstance(colName, str):
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "colName", "arg_type": type(colName).__name__},
+ )
return Column(UnresolvedRegex(colName))
colRegex.__doc__ = PySparkDataFrame.colRegex.__doc__
@@ -225,7 +231,13 @@ class DataFrame:
self.sparkSession,
)
else:
- raise TypeError("numPartitions should be an int or Column")
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={
+ "arg_name": "numPartitions",
+ "arg_type": type(numPartitions).__name__,
+ },
+ )
repartition.__doc__ = PySparkDataFrame.repartition.__doc__
@@ -277,8 +289,11 @@ class DataFrame:
repartitionByRange.__doc__ = PySparkDataFrame.repartitionByRange.__doc__
def dropDuplicates(self, subset: Optional[List[str]] = None) -> "DataFrame":
- if subset is not None and (not isinstance(subset, Iterable) or isinstance(subset, str)):
- raise TypeError("Parameter 'subset' must be a list of columns")
+ if subset is not None and not isinstance(subset, (list, tuple)):
+ raise PySparkTypeError(
+ error_class="NOT_LIST_OR_TUPLE",
+ message_parameters={"arg_name": "subset", "arg_type": type(subset).__name__},
+ )
if subset is None:
return DataFrame.withPlan(
@@ -558,7 +573,10 @@ class DataFrame:
def withColumnsRenamed(self, colsMap: Dict[str, str]) -> "DataFrame":
if not isinstance(colsMap, dict):
- raise TypeError("colsMap must be dict of existing column name and new column name.")
+ raise PySparkTypeError(
+ error_class="NOT_A_DICT",
+ message_parameters={"arg_name": "colsMap", "arg_type": type(colsMap).__name__},
+ )
return DataFrame.withPlan(plan.WithColumnsRenamed(self._plan, colsMap), self._session)
@@ -568,9 +586,15 @@ class DataFrame:
self, n: int = 20, truncate: Union[bool, int] = True, vertical: bool = False
) -> str:
if not isinstance(n, int) or isinstance(n, bool):
- raise TypeError("Parameter 'n' (number of rows) must be an int")
+ raise PySparkTypeError(
+ error_class="NOT_AN_INTEGER",
+ message_parameters={"arg_name": "n", "arg_type": type(n).__name__},
+ )
if not isinstance(vertical, bool):
- raise TypeError("Parameter 'vertical' must be a bool")
+ raise PySparkTypeError(
+ error_class="NOT_A_BOOLEAN",
+ message_parameters={"arg_name": "vertical", "arg_type": type(vertical).__name__},
+ )
_truncate: int = -1
if isinstance(truncate, bool) and truncate:
@@ -579,8 +603,12 @@ class DataFrame:
try:
_truncate = int(truncate)
except ValueError:
- raise TypeError(
- "Parameter 'truncate={}' should be either bool or int.".format(truncate)
+ raise PySparkTypeError(
+ error_class="NOT_A_BOOLEAN",
+ message_parameters={
+ "arg_name": "truncate",
+ "arg_type": type(truncate).__name__,
+ },
)
pdf = DataFrame.withPlan(
@@ -806,6 +834,11 @@ class DataFrame:
intersectAll.__doc__ = PySparkDataFrame.intersectAll.__doc__
def where(self, condition: Union[Column, str]) -> "DataFrame":
+ if not isinstance(condition, (str, Column)):
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "condition", "arg_type": type(condition).__name__},
+ )
return self.filter(condition)
where.__doc__ = PySparkDataFrame.where.__doc__
@@ -822,9 +855,9 @@ class DataFrame:
subset: Optional[Union[str, Tuple[str, ...], List[str]]] = None,
) -> "DataFrame":
if not isinstance(value, (float, int, str, bool, dict)):
- raise TypeError(
- f"value should be a float, int, string, bool or dict, "
- f"but got {type(value).__name__}"
+ raise PySparkTypeError(
+ error_class="NOT_BOOL_OR_DICT_OR_FLOAT_OR_INTEGER_OR_STRING",
+ message_parameters={"arg_name": "value", "arg_type": type(value).__name__},
)
if isinstance(value, dict):
if len(value) == 0:
@@ -853,9 +886,9 @@ class DataFrame:
)
_cols = list(subset)
else:
- raise TypeError(
- f"cols should be a str, tuple[str] or list[str], "
- f"but got {type(subset).__name__}"
+ raise PySparkTypeError(
+ error_class="NOT_LIST_OR_TUPLE",
+ message_parameters={"arg_name": "subset", "arg_type": type(subset).__name__},
)
if isinstance(value, dict):
@@ -909,9 +942,9 @@ class DataFrame:
)
_cols = list(subset)
else:
- raise TypeError(
- f"cols should be a str, tuple[str] or list[str], "
- f"but got {type(subset).__name__}"
+ raise PySparkTypeError(
+ error_class="NOT_LIST_OR_STRING_OR_TUPLE",
+ message_parameters={"arg_name": "subset", "arg_type": type(subset).__name__},
)
return DataFrame.withPlan(
@@ -935,7 +968,10 @@ class DataFrame:
if isinstance(to_replace, dict):
value = None
else:
- raise TypeError("value argument is required when to_replace is not a dictionary.")
+ raise PySparkTypeError(
+ error_class="ARGUMENT_REQUIRED",
+ message_parameters={"arg_name": "value", "condition": "`to_replace` is dict"},
+ )
# Helper functions
def all_of(types: Union[Type, Tuple[Type, ...]]) -> Callable[[Iterable], bool]:
@@ -960,9 +996,12 @@ class DataFrame:
# Validate input types
valid_types = (bool, float, int, str, list, tuple)
if not isinstance(to_replace, valid_types + (dict,)):
- raise TypeError(
- "to_replace should be a bool, float, int, string, list, tuple, or dict. "
- "Got {0}".format(type(to_replace))
+ raise PySparkTypeError(
+ error_class="NOT_BOOL_OR_DICT_OR_FLOAT_OR_INTEGER_OR_LIST_OR_STRING_OR_TUPLE",
+ message_parameters={
+ "arg_name": "to_replace",
+ "arg_type": type(to_replace).__name__,
+ },
)
if (
@@ -1057,9 +1096,15 @@ class DataFrame:
def cov(self, col1: str, col2: str) -> float:
if not isinstance(col1, str):
- raise TypeError("col1 should be a string.")
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "col1", "arg_type": type(col1).__name__},
+ )
if not isinstance(col2, str):
- raise TypeError("col2 should be a string.")
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "col2", "arg_type": type(col2).__name__},
+ )
pdf = DataFrame.withPlan(
plan.StatCov(child=self._plan, col1=col1, col2=col2),
session=self._session,
@@ -1178,12 +1223,26 @@ class DataFrame:
if isinstance(col, str):
col = Column(ColumnReference(col))
elif not isinstance(col, Column):
- raise TypeError("col must be a string or a column, but got %r" % type(col))
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "col", "arg_type": type(col).__name__},
+ )
if not isinstance(fractions, dict):
- raise TypeError("fractions must be a dict but got %r" % type(fractions))
+ raise PySparkTypeError(
+ error_class="NOT_A_DICT",
+ message_parameters={"arg_name": "fractions", "arg_type": type(fractions).__name__},
+ )
for k, v in fractions.items():
if not isinstance(k, (float, int, str)):
- raise TypeError("key must be float, int, or string, but got %r" % type(k))
+ raise PySparkTypeError(
+ error_class="DISALLOWED_TYPE_FOR_CONTAINER",
+ message_parameters={
+ "arg_name": "fractions",
+ "arg_type": type(fractions).__name__,
+ "allowed_types": "float, int, str",
+ "return_type": type(k).__name__,
+ },
+ )
fractions[k] = float(v)
seed = seed if seed is not None else random.randint(0, sys.maxsize)
return DataFrame.withPlan(
diff --git a/python/pyspark/sql/dataframe.py b/python/pyspark/sql/dataframe.py
index f45b0d70e99..f9d9c199faf 100644
--- a/python/pyspark/sql/dataframe.py
+++ b/python/pyspark/sql/dataframe.py
@@ -44,6 +44,7 @@ from py4j.java_gateway import JavaObject
from pyspark import copy_func, _NoValue
from pyspark._globals import _NoValueType
from pyspark.context import SparkContext
+from pyspark.errors import PySparkTypeError
from pyspark.rdd import (
RDD,
_load_from_socket,
@@ -692,11 +693,19 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
is_mode_case = extended is None and isinstance(mode, str)
if not (is_no_argument or is_extended_case or is_extended_as_mode or is_mode_case):
- argtypes = [str(type(arg)) for arg in [extended, mode] if arg is not None]
- raise TypeError(
- "extended (optional) and mode (optional) should be a string "
- "and bool; however, got [%s]." % ", ".join(argtypes)
- )
+ if (extended is not None) and (not isinstance(extended, (bool, str))):
+ raise PySparkTypeError(
+ error_class="NOT_BOOL_OR_STRING",
+ message_parameters={
+ "arg_name": "extended",
+ "arg_type": type(extended).__name__,
+ },
+ )
+ if (mode is not None) and (not isinstance(mode, str)):
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "mode", "arg_type": type(mode).__name__},
+ )
# Sets an explain mode depending on a given argument
if is_no_argument:
@@ -891,10 +900,16 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
"""
if not isinstance(n, int) or isinstance(n, bool):
- raise TypeError("Parameter 'n' (number of rows) must be an int")
+ raise PySparkTypeError(
+ error_class="NOT_AN_INTEGER",
+ message_parameters={"arg_name": "n", "arg_type": type(n).__name__},
+ )
if not isinstance(vertical, bool):
- raise TypeError("Parameter 'vertical' must be a bool")
+ raise PySparkTypeError(
+ error_class="NOT_A_BOOLEAN",
+ message_parameters={"arg_name": "vertical", "arg_type": type(vertical).__name__},
+ )
if isinstance(truncate, bool) and truncate:
print(self._jdf.showString(n, 20, vertical))
@@ -902,8 +917,12 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
try:
int_truncate = int(truncate)
except ValueError:
- raise TypeError(
- "Parameter 'truncate={}' should be either bool or int.".format(truncate)
+ raise PySparkTypeError(
+ error_class="NOT_A_BOOLEAN",
+ message_parameters={
+ "arg_name": "truncate",
+ "arg_type": type(truncate).__name__,
+ },
)
print(self._jdf.showString(n, int_truncate, vertical))
@@ -1084,9 +1103,18 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
>>> query.stop()
"""
if not eventTime or type(eventTime) is not str:
- raise TypeError("eventTime should be provided as a string")
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "eventTime", "arg_type": type(eventTime).__name__},
+ )
if not delayThreshold or type(delayThreshold) is not str:
- raise TypeError("delayThreshold should be provided as a string interval")
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={
+ "arg_name": "delayThreshold",
+ "arg_type": type(delayThreshold).__name__,
+ },
+ )
jdf = self._jdf.withWatermark(eventTime, delayThreshold)
return DataFrame(jdf, self.sparkSession)
@@ -1134,15 +1162,22 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
parameters = parameters[0] # type: ignore[assignment]
if not isinstance(name, str):
- raise TypeError("name should be provided as str, got {0}".format(type(name)))
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "name", "arg_type": type(name).__name__},
+ )
allowed_types = (str, list, float, int)
for p in parameters:
if not isinstance(p, allowed_types):
- raise TypeError(
- "all parameters should be in {0}, got {1} of type {2}".format(
- allowed_types, p, type(p)
- )
+ raise PySparkTypeError(
+ error_class="DISALLOWED_TYPE_FOR_CONTAINER",
+ message_parameters={
+ "arg_name": "parameters",
+ "arg_type": type(parameters).__name__,
+ "allowed_types": ", ".join(map(lambda x: x.__name__, allowed_types)),
+ "return_type": type(p).__name__,
+ },
)
jdf = self._jdf.hint(name, self._jseq(parameters))
@@ -1621,7 +1656,13 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
cols = (numPartitions,) + cols
return DataFrame(self._jdf.repartition(self._jcols(*cols)), self.sparkSession)
else:
- raise TypeError("numPartitions should be an int or Column")
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={
+ "arg_name": "numPartitions",
+ "arg_type": type(numPartitions).__name__,
+ },
+ )
@overload
def repartitionByRange(self, numPartitions: int, *cols: "ColumnOrName") -> "DataFrame":
@@ -1688,7 +1729,13 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
cols = (numPartitions,) + cols
return DataFrame(self._jdf.repartitionByRange(self._jcols(*cols)), self.sparkSession)
else:
- raise TypeError("numPartitions should be an int, string or Column")
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_INTEGER_OR_STRING",
+ message_parameters={
+ "arg_name": "numPartitions",
+ "arg_type": type(numPartitions).__name__,
+ },
+ )
def distinct(self) -> "DataFrame":
"""Returns a new :class:`DataFrame` containing the distinct rows in this :class:`DataFrame`.
@@ -1801,7 +1848,7 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
argtypes = [
str(type(arg)) for arg in [withReplacement, fraction, seed] if arg is not None
]
- raise TypeError(
+ raise PySparkTypeError(
"withReplacement (optional), fraction (required) and seed (optional)"
" should be a bool, float and number; however, "
"got [%s]." % ", ".join(argtypes)
@@ -1862,12 +1909,26 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
if isinstance(col, str):
col = Column(col)
elif not isinstance(col, Column):
- raise TypeError("col must be a string or a column, but got %r" % type(col))
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "col", "arg_type": type(col).__name__},
+ )
if not isinstance(fractions, dict):
- raise TypeError("fractions must be a dict but got %r" % type(fractions))
+ raise PySparkTypeError(
+ error_class="NOT_A_DICT",
+ message_parameters={"arg_name": "fractions", "arg_type": type(fractions).__name__},
+ )
for k, v in fractions.items():
if not isinstance(k, (float, int, str)):
- raise TypeError("key must be float, int, or string, but got %r" % type(k))
+ raise PySparkTypeError(
+ error_class="DISALLOWED_TYPE_FOR_CONTAINER",
+ message_parameters={
+ "arg_name": "fractions",
+ "arg_type": type(fractions).__name__,
+ "allowed_types": "float, int, str",
+ "return_type": type(k).__name__,
+ },
+ )
fractions[k] = float(v)
col = col._jc
seed = seed if seed is not None else random.randint(0, sys.maxsize)
@@ -1999,7 +2060,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
+----+
"""
if not isinstance(colName, str):
- raise TypeError("colName should be provided as string")
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "colName", "arg_type": type(colName).__name__},
+ )
jc = self._jdf.colRegex(colName)
return Column(jc)
@@ -2552,7 +2616,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
elif isinstance(ascending, list):
jcols = [jc if asc else jc.desc() for asc, jc in zip(ascending, jcols)]
else:
- raise TypeError("ascending can only be boolean or list, but got %s" % type(ascending))
+ raise PySparkTypeError(
+ error_class="NOT_BOOL_OR_LIST",
+ message_parameters={"arg_name": "ascending", "arg_type": type(ascending).__name__},
+ )
return self._jseq(jcols)
def describe(self, *cols: Union[str, List[str]]) -> "DataFrame":
@@ -2822,7 +2889,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
jc = self._jdf.apply(self.columns[item])
return Column(jc)
else:
- raise TypeError("unexpected item type: %s" % type(item))
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_FLOAT_OR_INTEGER_OR_LIST_OR_STRING",
+ message_parameters={"arg_name": "item", "arg_type": type(item).__name__},
+ )
def __getattr__(self, name: str) -> Column:
"""Returns the :class:`Column` denoted by ``name``.
@@ -3018,7 +3088,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
elif isinstance(condition, Column):
jdf = self._jdf.filter(condition._jc)
else:
- raise TypeError("condition should be string or Column")
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "condition", "arg_type": type(condition).__name__},
+ )
return DataFrame(jdf, self.sparkSession)
@overload
@@ -3796,7 +3869,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
+-----+---+------+
"""
if subset is not None and (not isinstance(subset, Iterable) or isinstance(subset, str)):
- raise TypeError("Parameter 'subset' must be a list of columns")
+ raise PySparkTypeError(
+ error_class="NOT_LIST_OR_TUPLE",
+ message_parameters={"arg_name": "subset", "arg_type": type(subset).__name__},
+ )
if subset is None:
jdf = self._jdf.dropDuplicates()
@@ -3860,7 +3936,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
elif isinstance(subset, str):
subset = [subset]
elif not isinstance(subset, (list, tuple)):
- raise TypeError("subset should be a list or tuple of column names")
+ raise PySparkTypeError(
+ error_class="NOT_LIST_OR_STRING_OR_TUPLE",
+ message_parameters={"arg_name": "subset", "arg_type": type(subset).__name__},
+ )
if thresh is None:
thresh = len(subset) if how == "any" else 1
@@ -3956,7 +4035,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
+---+------+-------+----+
"""
if not isinstance(value, (float, int, str, bool, dict)):
- raise TypeError("value should be a float, int, string, bool or dict")
+ raise PySparkTypeError(
+ error_class="NOT_BOOL_OR_DICT_OR_FLOAT_OR_INTEGER_OR_STRING",
+ message_parameters={"arg_name": "value", "arg_type": type(value).__name__},
+ )
# Note that bool validates isinstance(int), but we don't want to
# convert bools to floats
@@ -3972,7 +4054,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
if isinstance(subset, str):
subset = [subset]
elif not isinstance(subset, (list, tuple)):
- raise TypeError("subset should be a list or tuple of column names")
+ raise PySparkTypeError(
+ error_class="NOT_LIST_OR_TUPLE",
+ message_parameters={"arg_name": "subset", "arg_type": type(subset).__name__},
+ )
return DataFrame(self._jdf.na().fill(value, self._jseq(subset)), self.sparkSession)
@@ -4107,7 +4192,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
if isinstance(to_replace, dict):
value = None
else:
- raise TypeError("value argument is required when to_replace is not a dictionary.")
+ raise PySparkTypeError(
+ error_class="ARGUMENT_REQUIRED",
+ message_parameters={"arg_name": "value", "condition": "`to_replace` is dict"},
+ )
# Helper functions
def all_of(types: Union[Type, Tuple[Type, ...]]) -> Callable[[Iterable], bool]:
@@ -4132,9 +4220,12 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
# Validate input types
valid_types = (bool, float, int, str, list, tuple)
if not isinstance(to_replace, valid_types + (dict,)):
- raise TypeError(
- "to_replace should be a bool, float, int, string, list, tuple, or dict. "
- "Got {0}".format(type(to_replace))
+ raise PySparkTypeError(
+ error_class="NOT_BOOL_OR_DICT_OR_FLOAT_OR_INTEGER_OR_LIST_OR_STRING_OR_TUPLE",
+ message_parameters={
+ "arg_name": "to_replace",
+ "arg_type": type(to_replace).__name__,
+ },
)
if (
@@ -4142,7 +4233,7 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
and value is not None
and not isinstance(to_replace, dict)
):
- raise TypeError(
+ raise PySparkTypeError(
"If to_replace is not a dict, value should be "
"a bool, float, int, string, list, tuple or None. "
"Got {0}".format(type(value))
@@ -4156,9 +4247,9 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
)
if not (subset is None or isinstance(subset, (list, tuple, str))):
- raise TypeError(
- "subset should be a list or tuple of column names, "
- "column name or None. Got {0}".format(type(subset))
+ raise PySparkTypeError(
+ error_class="NOT_LIST_OR_STRING_OR_TUPLE",
+ message_parameters={"arg_name": "subset", "arg_type": type(subset).__name__},
)
# Reshape input arguments if necessary
@@ -4272,7 +4363,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
"""
if not isinstance(col, (str, list, tuple)):
- raise TypeError("col should be a string, list or tuple, but got %r" % type(col))
+ raise PySparkTypeError(
+ error_class="NOT_LIST_OR_STRING_OR_TUPLE",
+ message_parameters={"arg_name": "col", "arg_type": type(col).__name__},
+ )
isStr = isinstance(col, str)
@@ -4283,11 +4377,25 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
for c in col:
if not isinstance(c, str):
- raise TypeError("columns should be strings, but got %r" % type(c))
+ raise PySparkTypeError(
+ error_class="DISALLOWED_TYPE_FOR_CONTAINER",
+ message_parameters={
+ "arg_name": "col",
+ "arg_type": type(col).__name__,
+ "allowed_types": "str",
+ "return_type": type(c).__name__,
+ },
+ )
col = _to_list(self._sc, cast(List["ColumnOrName"], col))
if not isinstance(probabilities, (list, tuple)):
- raise TypeError("probabilities should be a list or tuple")
+ raise PySparkTypeError(
+ error_class="NOT_LIST_OR_TUPLE",
+ message_parameters={
+ "arg_name": "probabilities",
+ "arg_type": type(probabilities).__name__,
+ },
+ )
if isinstance(probabilities, tuple):
probabilities = list(probabilities)
for p in probabilities:
@@ -4296,7 +4404,13 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
probabilities = _to_list(self._sc, cast(List["ColumnOrName"], probabilities))
if not isinstance(relativeError, (float, int)):
- raise TypeError("relativeError should be numerical (float, int)")
+ raise PySparkTypeError(
+ error_class="NOT_FLOAT_OR_INTEGER",
+ message_parameters={
+ "arg_name": "relativeError",
+ "arg_type": type(relativeError).__name__,
+ },
+ )
if relativeError < 0:
raise ValueError("relativeError should be >= 0.")
relativeError = float(relativeError)
@@ -4338,9 +4452,15 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
"""
if not isinstance(col1, str):
- raise TypeError("col1 should be a string.")
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "col1", "arg_type": type(col1).__name__},
+ )
if not isinstance(col2, str):
- raise TypeError("col2 should be a string.")
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "col2", "arg_type": type(col2).__name__},
+ )
if not method:
method = "pearson"
if not method == "pearson":
@@ -4380,9 +4500,15 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
"""
if not isinstance(col1, str):
- raise TypeError("col1 should be a string.")
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "col1", "arg_type": type(col1).__name__},
+ )
if not isinstance(col2, str):
- raise TypeError("col2 should be a string.")
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "col2", "arg_type": type(col2).__name__},
+ )
return self._jdf.stat().cov(col1, col2)
def crosstab(self, col1: str, col2: str) -> "DataFrame":
@@ -4427,9 +4553,15 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
"""
if not isinstance(col1, str):
- raise TypeError("col1 should be a string.")
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "col1", "arg_type": type(col1).__name__},
+ )
if not isinstance(col2, str):
- raise TypeError("col2 should be a string.")
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "col2", "arg_type": type(col2).__name__},
+ )
return DataFrame(self._jdf.stat().crosstab(col1, col2), self.sparkSession)
def freqItems(
@@ -4476,7 +4608,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
if isinstance(cols, tuple):
cols = list(cols)
if not isinstance(cols, list):
- raise TypeError("cols must be a list or tuple of column names as strings.")
+ raise PySparkTypeError(
+ error_class="NOT_LIST_OR_TUPLE",
+ message_parameters={"arg_name": "cols", "arg_type": type(cols).__name__},
+ )
if not support:
support = 0.01
return DataFrame(
@@ -4523,7 +4658,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
colsMap = colsMap[0] # type: ignore[assignment]
if not isinstance(colsMap, dict):
- raise TypeError("colsMap must be dict of column name and column.")
+ raise PySparkTypeError(
+ error_class="NOT_A_DICT",
+ message_parameters={"arg_name": "colsMap", "arg_type": type(colsMap).__name__},
+ )
col_names = list(colsMap.keys())
cols = list(colsMap.values())
@@ -4577,7 +4715,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
+---+-----+----+
"""
if not isinstance(col, Column):
- raise TypeError("col should be Column")
+ raise PySparkTypeError(
+ error_class="NOT_A_COLUMN",
+ message_parameters={"arg_name": "col", "arg_type": type(col).__name__},
+ )
return DataFrame(self._jdf.withColumn(colName, col._jc), self.sparkSession)
def withColumnRenamed(self, existing: str, new: str) -> "DataFrame":
@@ -4653,7 +4794,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
+---+-----+----+----+
"""
if not isinstance(colsMap, dict):
- raise TypeError("colsMap must be dict of existing column name and new column name.")
+ raise PySparkTypeError(
+ error_class="NOT_A_DICT",
+ message_parameters={"arg_name": "colsMap", "arg_type": type(colsMap).__name__},
+ )
return DataFrame(self._jdf.withColumnsRenamed(colsMap), self.sparkSession)
@@ -4682,7 +4826,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
{'foo': 'bar'}
"""
if not isinstance(metadata, dict):
- raise TypeError("metadata should be a dict")
+ raise PySparkTypeError(
+ error_class="NOT_A_DICT",
+ message_parameters={"arg_name": "metadata", "arg_type": type(metadata).__name__},
+ )
sc = SparkContext._active_spark_context
assert sc is not None and sc._jvm is not None
jmeta = sc._jvm.org.apache.spark.sql.types.Metadata.fromJson(json.dumps(metadata))
@@ -4756,7 +4903,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
elif isinstance(col, Column):
jdf = self._jdf.drop(col._jc)
else:
- raise TypeError("col should be a string or a Column")
+ raise PySparkTypeError(
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "col", "arg_type": type(col).__name__},
+ )
else:
jcols = [_to_java_column(c) for c in cols]
first_column, *remaining_columns = jcols
@@ -4897,7 +5047,10 @@ class DataFrame(PandasMapOpsMixin, PandasConversionMixin):
True
"""
if not isinstance(other, DataFrame):
- raise TypeError("other parameter should be of DataFrame; however, got %s" % type(other))
+ raise PySparkTypeError(
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "other", "arg_type": type(other).__name__},
+ )
return self._jdf.sameSemantics(other._jdf)
def semanticHash(self) -> int:
diff --git a/python/pyspark/sql/tests/connect/test_connect_basic.py b/python/pyspark/sql/tests/connect/test_connect_basic.py
index 94aed9fcc30..061591e742c 100644
--- a/python/pyspark/sql/tests/connect/test_connect_basic.py
+++ b/python/pyspark/sql/tests/connect/test_connect_basic.py
@@ -22,6 +22,7 @@ import unittest
import shutil
import tempfile
+from pyspark.errors import PySparkTypeError
from pyspark.testing.sqlutils import SQLTestUtils
from pyspark.sql import SparkSession as PySparkSession, Row
from pyspark.sql.types import (
@@ -1751,9 +1752,20 @@ class SparkConnectBasicTests(SparkConnectSQLTestCase):
.toPandas(),
)
- with self.assertRaisesRegex(TypeError, "key must be float, int, or string"):
+ with self.assertRaises(PySparkTypeError) as pe:
cdf.stat.sampleBy(cdf.key, fractions={0: 0.1, None: 0.2}, seed=0)
+ self.check_error(
+ exception=pe.exception,
+ error_class="DISALLOWED_TYPE_FOR_CONTAINER",
+ message_parameters={
+ "arg_name": "fractions",
+ "arg_type": "dict",
+ "allowed_types": "float, int, str",
+ "return_type": "NoneType",
+ },
+ )
+
with self.assertRaises(SparkConnectException):
cdf.sampleBy(cdf.key, fractions={0: 0.1, 1: 1.2}, seed=0).show()
diff --git a/python/pyspark/sql/tests/test_dataframe.py b/python/pyspark/sql/tests/test_dataframe.py
index 0ba0649245c..d0fd39f4e05 100644
--- a/python/pyspark/sql/tests/test_dataframe.py
+++ b/python/pyspark/sql/tests/test_dataframe.py
@@ -48,6 +48,7 @@ from pyspark.errors import (
IllegalArgumentException,
SparkConnectException,
SparkConnectAnalysisException,
+ PySparkTypeError,
)
from pyspark.testing.sqlutils import (
ReusedSQLTestCase,
@@ -116,10 +117,15 @@ class DataFrameTestsMixin:
self.assertEqual(renamed_df2.columns, ["naam", "age"])
# negative test for incorrect type
- type_error_msg = "colsMap must be dict of existing column name and new column name."
- with self.assertRaisesRegex(TypeError, type_error_msg):
+ with self.assertRaises(PySparkTypeError) as pe:
df.withColumnsRenamed(("name", "x"))
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_A_DICT",
+ message_parameters={"arg_name": "colsMap", "arg_type": "tuple"},
+ )
+
def test_drop_duplicates(self):
# SPARK-36034 test that drop duplicates throws a type error when in correct type provided
df = self.spark.createDataFrame([("Alice", 50), ("Alice", 60)], ["name", "age"])
@@ -131,10 +137,15 @@ class DataFrameTestsMixin:
self.assertEqual(df.dropDuplicates(["name", "age"]).count(), 2)
- type_error_msg = "Parameter 'subset' must be a list of columns"
- with self.assertRaisesRegex(TypeError, type_error_msg):
+ with self.assertRaises(PySparkTypeError) as pe:
df.dropDuplicates("name")
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_LIST_OR_TUPLE",
+ message_parameters={"arg_name": "subset", "arg_type": "str"},
+ )
+
def test_dropna(self):
schema = StructType(
[
@@ -209,6 +220,15 @@ class DataFrameTestsMixin:
1,
)
+ with self.assertRaises(PySparkTypeError) as pe:
+ self.spark.createDataFrame([("Alice", 50, None)], schema).dropna(subset=10)
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_LIST_OR_STRING_OR_TUPLE",
+ message_parameters={"arg_name": "subset", "arg_type": "int"},
+ )
+
def test_fillna(self):
schema = StructType(
[
@@ -280,6 +300,24 @@ class DataFrameTestsMixin:
row = self.spark.createDataFrame([Row(a=None), Row(a=True)]).fillna({"a": True}).first()
self.assertEqual(row.a, True)
+ with self.assertRaises(PySparkTypeError) as pe:
+ self.spark.createDataFrame([Row(a=None), Row(a=True)]).fillna(["a", True])
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_BOOL_OR_DICT_OR_FLOAT_OR_INTEGER_OR_STRING",
+ message_parameters={"arg_name": "value", "arg_type": "list"},
+ )
+
+ with self.assertRaises(PySparkTypeError) as pe:
+ self.spark.createDataFrame([Row(a=None), Row(a=True)]).fillna(50, subset=10)
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_LIST_OR_TUPLE",
+ message_parameters={"arg_name": "subset", "arg_type": "int"},
+ )
+
def test_repartitionByRange_dataframe(self):
schema = StructType(
[
@@ -313,6 +351,15 @@ class DataFrameTestsMixin:
self.assertEqual(df5.rdd.first(), df2.rdd.first())
self.assertEqual(df5.rdd.take(3), df2.rdd.take(3))
+ with self.assertRaises(PySparkTypeError) as pe:
+ df1.repartitionByRange([10], "name", "age")
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_COLUMN_OR_INTEGER_OR_STRING",
+ message_parameters={"arg_name": "numPartitions", "arg_type": "list"},
+ )
+
def test_replace(self):
schema = StructType(
[
@@ -494,12 +541,23 @@ class DataFrameTestsMixin:
{"Alice": "Bob", 10: 20}
).first()
- with self.assertRaisesRegex(
- TypeError, "value argument is required when to_replace is not a dictionary."
- ):
- self.spark.createDataFrame([("Alice", 10, 80.0)], schema).replace(
- ["Alice", "Bob"]
- ).first()
+ with self.assertRaises(PySparkTypeError) as pe:
+ self.spark.createDataFrame([("Alice", 10, 80.0)], schema).replace(["Alice", "Bob"])
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="ARGUMENT_REQUIRED",
+ message_parameters={"arg_name": "value", "condition": "`to_replace` is dict"},
+ )
+
+ with self.assertRaises(PySparkTypeError) as pe:
+ self.spark.createDataFrame([("Alice", 10, 80.0)], schema).replace(lambda x: x + 1, 10)
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_BOOL_OR_DICT_OR_FLOAT_OR_INTEGER_OR_LIST_OR_STRING_OR_TUPLE",
+ message_parameters={"arg_name": "to_replace", "arg_type": "function"},
+ )
def test_with_column_with_existing_name(self):
keys = self.df.withColumn("key", self.df.key).select("key").collect()
@@ -1426,9 +1484,15 @@ class DataFrameTestsMixin:
def test_same_semantics_error(self):
with QuietTest(self.sc):
- with self.assertRaisesRegex(TypeError, "should be of DataFrame.*int"):
+ with self.assertRaises(PySparkTypeError) as pe:
self.spark.range(10).sameSemantics(1)
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "other", "arg_type": "int"},
+ )
+
def test_input_files(self):
tpath = tempfile.mkdtemp()
shutil.rmtree(tpath)
@@ -1456,13 +1520,33 @@ class DataFrameTestsMixin:
df.show(n=5, truncate="1", vertical=False)
df.show(n=5, truncate=1.5, vertical=False)
- with self.assertRaisesRegex(TypeError, "Parameter 'n'"):
+ with self.assertRaises(PySparkTypeError) as pe:
df.show(True)
- with self.assertRaisesRegex(TypeError, "Parameter 'vertical'"):
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_AN_INTEGER",
+ message_parameters={"arg_name": "n", "arg_type": "bool"},
+ )
+
+ with self.assertRaises(PySparkTypeError) as pe:
df.show(vertical="foo")
- with self.assertRaisesRegex(TypeError, "Parameter 'truncate=foo'"):
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_A_BOOLEAN",
+ message_parameters={"arg_name": "vertical", "arg_type": "str"},
+ )
+
+ with self.assertRaises(PySparkTypeError) as pe:
df.show(truncate="foo")
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_A_BOOLEAN",
+ message_parameters={"arg_name": "truncate", "arg_type": "str"},
+ )
+
@unittest.skipIf(
not have_pandas or not have_pyarrow,
cast(str, pandas_requirement_message or pyarrow_requirement_message),
@@ -1524,6 +1608,37 @@ class DataFrameTestsMixin:
lambda: df.to(schema5).count(),
)
+ def test_repartition(self):
+ df = self.spark.createDataFrame([(14, "Tom"), (23, "Alice"), (16, "Bob")], ["age", "name"])
+ with self.assertRaises(PySparkTypeError) as pe:
+ df.repartition([10], "name", "age").rdd.getNumPartitions()
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "numPartitions", "arg_type": "list"},
+ )
+
+ def test_colregex(self):
+ with self.assertRaises(PySparkTypeError) as pe:
+ self.spark.range(10).colRegex(10)
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "colName", "arg_type": "int"},
+ )
+
+ def test_where(self):
+ with self.assertRaises(PySparkTypeError) as pe:
+ self.spark.range(10).where(10)
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "condition", "arg_type": "int"},
+ )
+
class QueryExecutionListenerTests(unittest.TestCase, SQLTestUtils):
# These tests are separate because it uses 'spark.sql.queryExecutionListeners' which is
diff --git a/python/pyspark/sql/tests/test_functions.py b/python/pyspark/sql/tests/test_functions.py
index 4cbef4b5387..f8098d185af 100644
--- a/python/pyspark/sql/tests/test_functions.py
+++ b/python/pyspark/sql/tests/test_functions.py
@@ -213,11 +213,61 @@ class FunctionsTestsMixin:
sampled = df.stat.sampleBy("b", fractions={0: 0.5, 1: 0.5}, seed=0)
self.assertTrue(35 <= sampled.count() <= 36)
+ with self.assertRaises(PySparkTypeError) as pe:
+ df.sampleBy(10, fractions={0: 0.5, 1: 0.5})
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_COLUMN_OR_STRING",
+ message_parameters={"arg_name": "col", "arg_type": "int"},
+ )
+
+ with self.assertRaises(PySparkTypeError) as pe:
+ df.sampleBy("b", fractions=[0.5, 0.5])
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_A_DICT",
+ message_parameters={"arg_name": "fractions", "arg_type": "list"},
+ )
+
+ with self.assertRaises(PySparkTypeError) as pe:
+ df.sampleBy("b", fractions={None: 0.5, 1: 0.5})
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="DISALLOWED_TYPE_FOR_CONTAINER",
+ message_parameters={
+ "arg_name": "fractions",
+ "arg_type": "dict",
+ "allowed_types": "float, int, str",
+ "return_type": "NoneType",
+ },
+ )
+
def test_cov(self):
df = self.spark.createDataFrame([Row(a=i, b=2 * i) for i in range(10)])
cov = df.stat.cov("a", "b")
self.assertTrue(abs(cov - 55.0 / 3) < 1e-6)
+ with self.assertRaises(PySparkTypeError) as pe:
+ df.stat.cov(10, "b")
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "col1", "arg_type": "int"},
+ )
+
+ with self.assertRaises(PySparkTypeError) as pe:
+ df.stat.cov("a", True)
+
+ self.check_error(
+ exception=pe.exception,
+ error_class="NOT_A_STRING",
+ message_parameters={"arg_name": "col2", "arg_type": "bool"},
+ )
+
def test_crosstab(self):
df = self.spark.createDataFrame([Row(a=i % 3, b=i % 2) for i in range(1, 7)])
ct = df.stat.crosstab("a", "b").collect()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org