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/03/01 00:43:57 UTC
[spark] branch branch-3.4 updated: [SPARK-42624][PYTHON][TESTS] Reorganize imports in test_functions
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 93612f4fa11 [SPARK-42624][PYTHON][TESTS] Reorganize imports in test_functions
93612f4fa11 is described below
commit 93612f4fa111a96109af8ca9077e64bc715c172a
Author: Takuya UESHIN <ue...@databricks.com>
AuthorDate: Wed Mar 1 09:43:30 2023 +0900
[SPARK-42624][PYTHON][TESTS] Reorganize imports in test_functions
### What changes were proposed in this pull request?
Reorganizes imports in `python/pyspark/sql/tests/test_functions.py`.
### Why are the changes needed?
Currently imports in the file `python/pyspark/sql/tests/test_functions.py` is not organized well.
There are individual imports in test functions, or imports by function names or module names, etc.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Modified tests.
Closes #40223 from ueshin/issues/SPARK-42624/imports.
Authored-by: Takuya UESHIN <ue...@databricks.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
(cherry picked from commit 8dbc47f1b987249df398435fc86cdb86edaee8c8)
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../sql/tests/connect/test_parity_functions.py | 3 +-
python/pyspark/sql/tests/test_functions.py | 487 ++++++++-------------
2 files changed, 185 insertions(+), 305 deletions(-)
diff --git a/python/pyspark/sql/tests/connect/test_parity_functions.py b/python/pyspark/sql/tests/connect/test_parity_functions.py
index 747f9a1b287..35ddf96e1fb 100644
--- a/python/pyspark/sql/tests/connect/test_parity_functions.py
+++ b/python/pyspark/sql/tests/connect/test_parity_functions.py
@@ -18,6 +18,7 @@
import unittest
from pyspark.errors.exceptions.connect import SparkConnectException
+from pyspark.sql.connect.column import Column
from pyspark.sql.tests.test_functions import FunctionsTestsMixin
from pyspark.testing.connectutils import ReusedConnectTestCase
@@ -42,8 +43,6 @@ class FunctionsParityTests(FunctionsTestsMixin, ReusedConnectTestCase):
self.check_raise_error(SparkConnectException)
def test_sorting_functions_with_column(self):
- from pyspark.sql.connect.column import Column
-
self.check_sorting_functions_with_column(Column)
diff --git a/python/pyspark/sql/tests/test_functions.py b/python/pyspark/sql/tests/test_functions.py
index 44f1b9a4d13..38de87b0e72 100644
--- a/python/pyspark/sql/tests/test_functions.py
+++ b/python/pyspark/sql/tests/test_functions.py
@@ -15,70 +15,20 @@
# limitations under the License.
#
-import datetime
-import io
from contextlib import redirect_stdout
+import datetime
from inspect import getmembers, isfunction
+import io
from itertools import chain
-import re
import math
+import re
import unittest
from py4j.protocol import Py4JJavaError
+
from pyspark.errors import PySparkTypeError, PySparkValueError
-from pyspark.sql import Row, Window, types
-from pyspark.sql.functions import (
- udf,
- input_file_name,
- col,
- percentile_approx,
- lit,
- assert_true,
- sum_distinct,
- sumDistinct,
- shiftleft,
- shiftLeft,
- shiftRight,
- shiftright,
- shiftrightunsigned,
- shiftRightUnsigned,
- octet_length,
- bit_length,
- sec,
- csc,
- cot,
- make_date,
- date_add,
- date_sub,
- add_months,
- array_repeat,
- arrays_zip,
- size,
- slice,
- least,
- regexp_replace,
- atan2,
- hypot,
- pow,
- pmod,
- map_from_arrays,
- map_contains_key,
- map_keys,
- map_values,
- map_entries,
- map_concat,
- map_from_entries,
- expr,
- schema_of_json,
- schema_of_csv,
- from_csv,
- greatest,
- when,
- window,
- session_window,
- bucket,
-)
-from pyspark.sql import functions
+from pyspark.sql import Row, Window, functions as F, types
+from pyspark.sql.column import Column
from pyspark.testing.sqlutils import ReusedSQLTestCase, SQLTestUtils
from pyspark.testing.utils import have_numpy
@@ -100,7 +50,7 @@ class FunctionsTestsMixin:
# jvm_excluded_fn needs to be updated.
jvm_fn_set = {name for (name, value) in getmembers(self.sc._jvm.functions)}
- py_fn_set = {name for (name, value) in getmembers(functions, isfunction) if name[0] != "_"}
+ py_fn_set = {name for (name, value) in getmembers(F, isfunction) if name[0] != "_"}
# Functions on the JVM side we do not expect to be available in python because they are
# depreciated, irrelevant to python, or have equivalents.
@@ -134,8 +84,6 @@ class FunctionsTestsMixin:
)
def test_explode(self):
- from pyspark.sql.functions import explode, explode_outer, posexplode_outer
-
d = [
Row(a=1, intlist=[1, 2, 3], mapfield={"a": "b"}),
Row(a=1, intlist=[], mapfield={}),
@@ -143,30 +91,28 @@ class FunctionsTestsMixin:
]
data = self.spark.createDataFrame(d)
- result = data.select(explode(data.intlist).alias("a")).select("a").collect()
+ result = data.select(F.explode(data.intlist).alias("a")).select("a").collect()
self.assertEqual(result[0][0], 1)
self.assertEqual(result[1][0], 2)
self.assertEqual(result[2][0], 3)
- result = data.select(explode(data.mapfield).alias("a", "b")).select("a", "b").collect()
+ result = data.select(F.explode(data.mapfield).alias("a", "b")).select("a", "b").collect()
self.assertEqual(result[0][0], "a")
self.assertEqual(result[0][1], "b")
- result = [tuple(x) for x in data.select(posexplode_outer("intlist")).collect()]
+ result = [tuple(x) for x in data.select(F.posexplode_outer("intlist")).collect()]
self.assertEqual(result, [(0, 1), (1, 2), (2, 3), (None, None), (None, None)])
- result = [tuple(x) for x in data.select(posexplode_outer("mapfield")).collect()]
+ result = [tuple(x) for x in data.select(F.posexplode_outer("mapfield")).collect()]
self.assertEqual(result, [(0, "a", "b"), (None, None, None), (None, None, None)])
- result = [x[0] for x in data.select(explode_outer("intlist")).collect()]
+ result = [x[0] for x in data.select(F.explode_outer("intlist")).collect()]
self.assertEqual(result, [1, 2, 3, None, None])
- result = [tuple(x) for x in data.select(explode_outer("mapfield")).collect()]
+ result = [tuple(x) for x in data.select(F.explode_outer("mapfield")).collect()]
self.assertEqual(result, [("a", "b"), (None, None), (None, None)])
def test_inline(self):
- from pyspark.sql.functions import inline, inline_outer
-
d = [
Row(structlist=[Row(b=1, c=2), Row(b=3, c=4)]),
Row(structlist=[Row(b=None, c=5), None]),
@@ -174,10 +120,10 @@ class FunctionsTestsMixin:
]
data = self.spark.createDataFrame(d)
- result = [tuple(x) for x in data.select(inline(data.structlist)).collect()]
+ result = [tuple(x) for x in data.select(F.inline(data.structlist)).collect()]
self.assertEqual(result, [(1, 2), (3, 4), (None, 5), (None, None)])
- result = [tuple(x) for x in data.select(inline_outer(data.structlist)).collect()]
+ result = [tuple(x) for x in data.select(F.inline_outer(data.structlist)).collect()]
self.assertEqual(result, [(1, 2), (3, 4), (None, 5), (None, None), (None, None)])
def test_basic_functions(self):
@@ -202,8 +148,6 @@ class FunctionsTestsMixin:
df.collect()
def test_corr(self):
- import math
-
df = self.spark.createDataFrame([Row(a=i, b=math.sqrt(i)) for i in range(10)])
corr = df.stat.corr("a", "b")
self.assertTrue(abs(corr - 0.95734012) < 1e-6)
@@ -279,42 +223,37 @@ class FunctionsTestsMixin:
def test_math_functions(self):
df = self.spark.createDataFrame([Row(a=i, b=2 * i) for i in range(10)])
- from pyspark.sql import functions
SQLTestUtils.assert_close(
- [math.cos(i) for i in range(10)], df.select(functions.cos(df.a)).collect()
- )
- SQLTestUtils.assert_close(
- [math.cos(i) for i in range(10)], df.select(functions.cos("a")).collect()
+ [math.cos(i) for i in range(10)], df.select(F.cos(df.a)).collect()
)
+ SQLTestUtils.assert_close([math.cos(i) for i in range(10)], df.select(F.cos("a")).collect())
SQLTestUtils.assert_close(
- [math.sin(i) for i in range(10)], df.select(functions.sin(df.a)).collect()
+ [math.sin(i) for i in range(10)], df.select(F.sin(df.a)).collect()
)
SQLTestUtils.assert_close(
- [math.sin(i) for i in range(10)], df.select(functions.sin(df["a"])).collect()
+ [math.sin(i) for i in range(10)], df.select(F.sin(df["a"])).collect()
)
SQLTestUtils.assert_close(
- [math.pow(i, 2 * i) for i in range(10)], df.select(functions.pow(df.a, df.b)).collect()
+ [math.pow(i, 2 * i) for i in range(10)], df.select(F.pow(df.a, df.b)).collect()
)
SQLTestUtils.assert_close(
- [math.pow(i, 2) for i in range(10)], df.select(functions.pow(df.a, 2)).collect()
+ [math.pow(i, 2) for i in range(10)], df.select(F.pow(df.a, 2)).collect()
)
SQLTestUtils.assert_close(
- [math.pow(i, 2) for i in range(10)], df.select(functions.pow(df.a, 2.0)).collect()
+ [math.pow(i, 2) for i in range(10)], df.select(F.pow(df.a, 2.0)).collect()
)
SQLTestUtils.assert_close(
- [math.hypot(i, 2 * i) for i in range(10)],
- df.select(functions.hypot(df.a, df.b)).collect(),
+ [math.hypot(i, 2 * i) for i in range(10)], df.select(F.hypot(df.a, df.b)).collect()
)
SQLTestUtils.assert_close(
- [math.hypot(i, 2 * i) for i in range(10)],
- df.select(functions.hypot("a", "b")).collect(),
+ [math.hypot(i, 2 * i) for i in range(10)], df.select(F.hypot("a", "b")).collect()
)
SQLTestUtils.assert_close(
- [math.hypot(i, 2) for i in range(10)], df.select(functions.hypot("a", 2)).collect()
+ [math.hypot(i, 2) for i in range(10)], df.select(F.hypot("a", 2)).collect()
)
SQLTestUtils.assert_close(
- [math.hypot(i, 2) for i in range(10)], df.select(functions.hypot(df.a, 2)).collect()
+ [math.hypot(i, 2) for i in range(10)], df.select(F.hypot(df.a, 2)).collect()
)
def test_inverse_trig_functions(self):
@@ -336,11 +275,9 @@ class FunctionsTestsMixin:
df.select(inv(trig(df.b))).collect(),
)
- from pyspark.sql import functions
-
- check(functions.cosh, functions.acosh, y_axis_symmetrical=True)
- check(functions.sinh, functions.asinh, y_axis_symmetrical=False)
- check(functions.tanh, functions.atanh, y_axis_symmetrical=False)
+ check(F.cosh, F.acosh, y_axis_symmetrical=True)
+ check(F.sinh, F.asinh, y_axis_symmetrical=False)
+ check(F.tanh, F.atanh, y_axis_symmetrical=False)
def test_reciprocal_trig_functions(self):
# SPARK-36683: Tests for reciprocal trig functions (SEC, CSC and COT)
@@ -360,34 +297,37 @@ class FunctionsTestsMixin:
def to_reciprocal_trig(func):
return [1.0 / func(i) if func(i) != 0 else math.inf for i in lst]
- SQLTestUtils.assert_close(to_reciprocal_trig(math.cos), df.select(sec(df.value)).collect())
- SQLTestUtils.assert_close(to_reciprocal_trig(math.sin), df.select(csc(df.value)).collect())
- SQLTestUtils.assert_close(to_reciprocal_trig(math.tan), df.select(cot(df.value)).collect())
+ SQLTestUtils.assert_close(
+ to_reciprocal_trig(math.cos), df.select(F.sec(df.value)).collect()
+ )
+ SQLTestUtils.assert_close(
+ to_reciprocal_trig(math.sin), df.select(F.csc(df.value)).collect()
+ )
+ SQLTestUtils.assert_close(
+ to_reciprocal_trig(math.tan), df.select(F.cot(df.value)).collect()
+ )
def test_rand_functions(self):
df = self.spark.createDataFrame([Row(key=i, value=str(i)) for i in range(100)])
- rnd = df.select("key", functions.rand()).collect()
+ rnd = df.select("key", F.rand()).collect()
for row in rnd:
assert row[1] >= 0.0 and row[1] <= 1.0, "got: %s" % row[1]
- rndn = df.select("key", functions.randn(5)).collect()
+ rndn = df.select("key", F.randn(5)).collect()
for row in rndn:
assert row[1] >= -4.0 and row[1] <= 4.0, "got: %s" % row[1]
# If the specified seed is 0, we should use it.
# https://issues.apache.org/jira/browse/SPARK-9691
- rnd1 = df.select("key", functions.rand(0)).collect()
- rnd2 = df.select("key", functions.rand(0)).collect()
+ rnd1 = df.select("key", F.rand(0)).collect()
+ rnd2 = df.select("key", F.rand(0)).collect()
self.assertEqual(sorted(rnd1), sorted(rnd2))
- rndn1 = df.select("key", functions.randn(0)).collect()
- rndn2 = df.select("key", functions.randn(0)).collect()
+ rndn1 = df.select("key", F.randn(0)).collect()
+ rndn2 = df.select("key", F.randn(0)).collect()
self.assertEqual(sorted(rndn1), sorted(rndn2))
def test_string_functions(self):
- from pyspark.sql import functions
- from pyspark.sql.functions import col, lit
-
string_functions = [
"upper",
"lower",
@@ -401,7 +341,7 @@ class FunctionsTestsMixin:
df = self.spark.createDataFrame([["nick"]], schema=["name"])
with self.assertRaises(PySparkTypeError) as pe:
- df.select(col("name").substr(0, lit(1)))
+ df.select(F.col("name").substr(0, F.lit(1)))
self.check_error(
exception=pe.exception,
@@ -416,27 +356,25 @@ class FunctionsTestsMixin:
for name in string_functions:
self.assertEqual(
- df.select(getattr(functions, name)("name")).first()[0],
- df.select(getattr(functions, name)(col("name"))).first()[0],
+ df.select(getattr(F, name)("name")).first()[0],
+ df.select(getattr(F, name)(F.col("name"))).first()[0],
)
def test_octet_length_function(self):
# SPARK-36751: add octet length api for python
df = self.spark.createDataFrame([("cat",), ("\U0001F408",)], ["cat"])
- actual = df.select(octet_length("cat")).collect()
+ actual = df.select(F.octet_length("cat")).collect()
self.assertEqual([Row(3), Row(4)], actual)
def test_bit_length_function(self):
# SPARK-36751: add bit length api for python
df = self.spark.createDataFrame([("cat",), ("\U0001F408",)], ["cat"])
- actual = df.select(bit_length("cat")).collect()
+ actual = df.select(F.bit_length("cat")).collect()
self.assertEqual([Row(24), Row(32)], actual)
def test_array_contains_function(self):
- from pyspark.sql.functions import array_contains
-
df = self.spark.createDataFrame([(["1", "2", "3"],), ([],)], ["data"])
- actual = df.select(array_contains(df.data, "1").alias("b")).collect()
+ actual = df.select(F.array_contains(df.data, "1").alias("b")).collect()
self.assertEqual([Row(b=True), Row(b=False)], actual)
def test_between_function(self):
@@ -448,11 +386,9 @@ class FunctionsTestsMixin:
)
def test_dayofweek(self):
- from pyspark.sql.functions import dayofweek
-
dt = datetime.datetime(2017, 11, 6)
df = self.spark.createDataFrame([Row(date=dt)])
- row = df.select(dayofweek(df.date)).first()
+ row = df.select(F.dayofweek(df.date)).first()
self.assertEqual(row[0], 2)
# Test added for SPARK-37738; change Python API to accept both col & int as input
@@ -466,9 +402,9 @@ class FunctionsTestsMixin:
self.assertTrue(
all(
df.select(
- date_add(df.date, df.add) == datetime.date(2021, 12, 29),
- date_add(df.date, "add") == datetime.date(2021, 12, 29),
- date_add(df.date, 3) == datetime.date(2021, 12, 30),
+ F.date_add(df.date, df.add) == datetime.date(2021, 12, 29),
+ F.date_add(df.date, "add") == datetime.date(2021, 12, 29),
+ F.date_add(df.date, 3) == datetime.date(2021, 12, 30),
).first()
)
)
@@ -484,9 +420,9 @@ class FunctionsTestsMixin:
self.assertTrue(
all(
df.select(
- date_sub(df.date, df.sub) == datetime.date(2021, 12, 25),
- date_sub(df.date, "sub") == datetime.date(2021, 12, 25),
- date_sub(df.date, 3) == datetime.date(2021, 12, 24),
+ F.date_sub(df.date, df.sub) == datetime.date(2021, 12, 25),
+ F.date_sub(df.date, "sub") == datetime.date(2021, 12, 25),
+ F.date_sub(df.date, 3) == datetime.date(2021, 12, 24),
).first()
)
)
@@ -502,9 +438,9 @@ class FunctionsTestsMixin:
self.assertTrue(
all(
df.select(
- add_months(df.date, df.add) == datetime.date(2022, 2, 27),
- add_months(df.date, "add") == datetime.date(2022, 2, 27),
- add_months(df.date, 3) == datetime.date(2022, 3, 27),
+ F.add_months(df.date, df.add) == datetime.date(2022, 2, 27),
+ F.add_months(df.date, "add") == datetime.date(2022, 2, 27),
+ F.add_months(df.date, 3) == datetime.date(2022, 3, 27),
).first()
)
)
@@ -512,49 +448,43 @@ class FunctionsTestsMixin:
def test_make_date(self):
# SPARK-36554: expose make_date expression
df = self.spark.createDataFrame([(2020, 6, 26)], ["Y", "M", "D"])
- row_from_col = df.select(make_date(df.Y, df.M, df.D)).first()
+ row_from_col = df.select(F.make_date(df.Y, df.M, df.D)).first()
self.assertEqual(row_from_col[0], datetime.date(2020, 6, 26))
- row_from_name = df.select(make_date("Y", "M", "D")).first()
+ row_from_name = df.select(F.make_date("Y", "M", "D")).first()
self.assertEqual(row_from_name[0], datetime.date(2020, 6, 26))
def test_expr(self):
- from pyspark.sql import functions
-
row = Row(a="length string", b=75)
df = self.spark.createDataFrame([row])
- result = df.select(functions.expr("length(a)")).collect()[0].asDict()
+ result = df.select(F.expr("length(a)")).collect()[0].asDict()
self.assertEqual(13, result["length(a)"])
# add test for SPARK-10577 (test broadcast join hint)
def test_functions_broadcast(self):
- from pyspark.sql.functions import broadcast
-
df1 = self.spark.createDataFrame([(1, "1"), (2, "2")], ("key", "value"))
df2 = self.spark.createDataFrame([(1, "1"), (2, "2")], ("key", "value"))
# equijoin - should be converted into broadcast join
with io.StringIO() as buf, redirect_stdout(buf):
- df1.join(broadcast(df2), "key").explain(True)
+ df1.join(F.broadcast(df2), "key").explain(True)
self.assertGreaterEqual(buf.getvalue().count("Broadcast"), 1)
# no join key -- should not be a broadcast join
with io.StringIO() as buf, redirect_stdout(buf):
- df1.crossJoin(broadcast(df2)).explain(True)
+ df1.crossJoin(F.broadcast(df2)).explain(True)
self.assertGreaterEqual(buf.getvalue().count("Broadcast"), 1)
# planner should not crash without a join
- broadcast(df1).explain(True)
+ F.broadcast(df1).explain(True)
def test_first_last_ignorenulls(self):
- from pyspark.sql import functions
-
df = self.spark.range(0, 100)
- df2 = df.select(functions.when(df.id % 3 == 0, None).otherwise(df.id).alias("id"))
+ df2 = df.select(F.when(df.id % 3 == 0, None).otherwise(df.id).alias("id"))
df3 = df2.select(
- functions.first(df2.id, False).alias("a"),
- functions.first(df2.id, True).alias("b"),
- functions.last(df2.id, False).alias("c"),
- functions.last(df2.id, True).alias("d"),
+ F.first(df2.id, False).alias("a"),
+ F.first(df2.id, True).alias("b"),
+ F.last(df2.id, False).alias("c"),
+ F.last(df2.id, True).alias("d"),
)
self.assertEqual([Row(a=None, b=1, c=None, d=98)], df3.collect())
@@ -588,20 +518,11 @@ class FunctionsTestsMixin:
self.assertRaises(TypeError, lambda: df.stat.approxQuantile(["a", 123], [0.1, 0.9], 0.1))
def test_sorting_functions_with_column(self):
- from pyspark.sql.column import Column
-
self.check_sorting_functions_with_column(Column)
def check_sorting_functions_with_column(self, tpe):
- from pyspark.sql import functions
-
- funs = [
- functions.asc_nulls_first,
- functions.asc_nulls_last,
- functions.desc_nulls_first,
- functions.desc_nulls_last,
- ]
- exprs = [col("x"), "x"]
+ funs = [F.asc_nulls_first, F.asc_nulls_last, F.desc_nulls_first, F.desc_nulls_last]
+ exprs = [F.col("x"), "x"]
for fun in funs:
for _expr in exprs:
@@ -610,44 +531,42 @@ class FunctionsTestsMixin:
self.assertIn(f"""'x {fun.__name__.replace("_", " ").upper()}'""", str(res))
for _expr in exprs:
- res = functions.asc(_expr)
+ res = F.asc(_expr)
self.assertIsInstance(res, tpe)
self.assertIn("""'x ASC NULLS FIRST'""", str(res))
for _expr in exprs:
- res = functions.desc(_expr)
+ res = F.desc(_expr)
self.assertIsInstance(res, tpe)
self.assertIn("""'x DESC NULLS LAST'""", str(res))
def test_sort_with_nulls_order(self):
- from pyspark.sql import functions
-
df = self.spark.createDataFrame(
[("Tom", 80), (None, 60), ("Alice", 50)], ["name", "height"]
)
self.assertEqual(
- df.select(df.name).orderBy(functions.asc_nulls_first("name")).collect(),
+ df.select(df.name).orderBy(F.asc_nulls_first("name")).collect(),
[Row(name=None), Row(name="Alice"), Row(name="Tom")],
)
self.assertEqual(
- df.select(df.name).orderBy(functions.asc_nulls_last("name")).collect(),
+ df.select(df.name).orderBy(F.asc_nulls_last("name")).collect(),
[Row(name="Alice"), Row(name="Tom"), Row(name=None)],
)
self.assertEqual(
- df.select(df.name).orderBy(functions.desc_nulls_first("name")).collect(),
+ df.select(df.name).orderBy(F.desc_nulls_first("name")).collect(),
[Row(name=None), Row(name="Tom"), Row(name="Alice")],
)
self.assertEqual(
- df.select(df.name).orderBy(functions.desc_nulls_last("name")).collect(),
+ df.select(df.name).orderBy(F.desc_nulls_last("name")).collect(),
[Row(name="Tom"), Row(name="Alice"), Row(name=None)],
)
def test_input_file_name_reset_for_rdd(self):
rdd = self.sc.textFile("python/test_support/hello/hello.txt").map(lambda x: {"data": x})
df = self.spark.createDataFrame(rdd, "data STRING")
- df.select(input_file_name().alias("file")).collect()
+ df.select(F.input_file_name().alias("file")).collect()
- non_file_df = self.spark.range(100).select(input_file_name())
+ non_file_df = self.spark.range(100).select(F.input_file_name())
results = non_file_df.collect()
self.assertTrue(len(results) == 100)
@@ -675,43 +594,37 @@ class FunctionsTestsMixin:
)
expected = [Row(sliced=[2, 3]), Row(sliced=[5])]
- self.assertTrue(
- all(
- [
- df.select(slice(df.x, 2, 2).alias("sliced")).collect() == expected,
- df.select(slice(df.x, lit(2), lit(2)).alias("sliced")).collect() == expected,
- df.select(slice("x", "index", "len").alias("sliced")).collect() == expected,
- ]
- )
+ self.assertEqual(df.select(F.slice(df.x, 2, 2).alias("sliced")).collect(), expected)
+ self.assertEqual(
+ df.select(F.slice(df.x, F.lit(2), F.lit(2)).alias("sliced")).collect(), expected
+ )
+ self.assertEqual(
+ df.select(F.slice("x", "index", "len").alias("sliced")).collect(), expected
)
self.assertEqual(
- df.select(slice(df.x, size(df.x) - 1, lit(1)).alias("sliced")).collect(),
+ df.select(F.slice(df.x, F.size(df.x) - 1, F.lit(1)).alias("sliced")).collect(),
[Row(sliced=[2]), Row(sliced=[4])],
)
self.assertEqual(
- df.select(slice(df.x, lit(1), size(df.x) - 1).alias("sliced")).collect(),
+ df.select(F.slice(df.x, F.lit(1), F.size(df.x) - 1).alias("sliced")).collect(),
[Row(sliced=[1, 2]), Row(sliced=[4])],
)
def test_array_repeat(self):
df = self.spark.range(1)
- df = df.withColumn("repeat_n", lit(3))
+ df = df.withColumn("repeat_n", F.lit(3))
expected = [Row(val=[0, 0, 0])]
- self.assertTrue(
- all(
- [
- df.select(array_repeat("id", 3).alias("val")).collect() == expected,
- df.select(array_repeat("id", lit(3)).alias("val")).collect() == expected,
- df.select(array_repeat("id", "repeat_n").alias("val")).collect() == expected,
- ]
- )
+ self.assertEqual(df.select(F.array_repeat("id", 3).alias("val")).collect(), expected)
+ self.assertEqual(df.select(F.array_repeat("id", F.lit(3)).alias("val")).collect(), expected)
+ self.assertEqual(
+ df.select(F.array_repeat("id", "repeat_n").alias("val")).collect(), expected
)
def test_input_file_name_udf(self):
df = self.spark.read.text("python/test_support/hello/hello.txt")
- df = df.select(udf(lambda x: x)("value"), input_file_name().alias("file"))
+ df = df.select(F.udf(lambda x: x)("value"), F.input_file_name().alias("file"))
file_name = df.collect()[0].file
self.assertTrue("python/test_support/hello/hello.txt" in file_name)
@@ -719,18 +632,14 @@ class FunctionsTestsMixin:
df = self.spark.createDataFrame([(1, 4, 3)], ["a", "b", "c"])
expected = [Row(least=1)]
- self.assertTrue(
- all(
- [
- df.select(least(df.a, df.b, df.c).alias("least")).collect() == expected,
- df.select(least(lit(3), lit(5), lit(1)).alias("least")).collect() == expected,
- df.select(least("a", "b", "c").alias("least")).collect() == expected,
- ]
- )
+ self.assertEqual(df.select(F.least(df.a, df.b, df.c).alias("least")).collect(), expected)
+ self.assertEqual(
+ df.select(F.least(F.lit(3), F.lit(5), F.lit(1)).alias("least")).collect(), expected
)
+ self.assertEqual(df.select(F.least("a", "b", "c").alias("least")).collect(), expected)
with self.assertRaises(PySparkValueError) as pe:
- df.select(least(df.a).alias("least")).collect()
+ df.select(F.least(df.a).alias("least")).collect()
self.check_error(
exception=pe.exception,
@@ -739,21 +648,17 @@ class FunctionsTestsMixin:
)
def test_overlay(self):
- from pyspark.sql.functions import col, lit, overlay
- from itertools import chain
- import re
-
actual = list(
chain.from_iterable(
[
re.findall("(overlay\\(.*\\))", str(x))
for x in [
- overlay(col("foo"), col("bar"), 1),
- overlay("x", "y", 3),
- overlay(col("x"), col("y"), 1, 3),
- overlay("x", "y", 2, 5),
- overlay("x", "y", lit(11)),
- overlay("x", "y", lit(2), lit(5)),
+ F.overlay(F.col("foo"), F.col("bar"), 1),
+ F.overlay("x", "y", 3),
+ F.overlay(F.col("x"), F.col("y"), 1, 3),
+ F.overlay("x", "y", 2, 5),
+ F.overlay("x", "y", F.lit(11)),
+ F.overlay("x", "y", F.lit(2), F.lit(5)),
]
]
)
@@ -773,18 +678,14 @@ class FunctionsTestsMixin:
df = self.spark.createDataFrame([("SPARK_SQL", "CORE", 7, 0)], ("x", "y", "pos", "len"))
exp = [Row(ol="SPARK_CORESQL")]
- self.assertTrue(
- all(
- [
- df.select(overlay(df.x, df.y, 7, 0).alias("ol")).collect() == exp,
- df.select(overlay(df.x, df.y, lit(7), lit(0)).alias("ol")).collect() == exp,
- df.select(overlay("x", "y", "pos", "len").alias("ol")).collect() == exp,
- ]
- )
+ self.assertEqual(df.select(F.overlay(df.x, df.y, 7, 0).alias("ol")).collect(), exp)
+ self.assertEqual(
+ df.select(F.overlay(df.x, df.y, F.lit(7), F.lit(0)).alias("ol")).collect(), exp
)
+ self.assertEqual(df.select(F.overlay("x", "y", "pos", "len").alias("ol")).collect(), exp)
with self.assertRaises(PySparkTypeError) as pe:
- df.select(overlay(df.x, df.y, 7.5, 0).alias("ol")).collect()
+ df.select(F.overlay(df.x, df.y, 7.5, 0).alias("ol")).collect()
self.check_error(
exception=pe.exception,
@@ -793,7 +694,7 @@ class FunctionsTestsMixin:
)
with self.assertRaises(PySparkTypeError) as pe:
- df.select(overlay(df.x, df.y, 7, 0.5).alias("ol")).collect()
+ df.select(F.overlay(df.x, df.y, 7, 0.5).alias("ol")).collect()
self.check_error(
exception=pe.exception,
@@ -807,12 +708,12 @@ class FunctionsTestsMixin:
[
re.findall("(percentile_approx\\(.*\\))", str(x))
for x in [
- percentile_approx(col("foo"), lit(0.5)),
- percentile_approx(col("bar"), 0.25, 42),
- percentile_approx(col("bar"), [0.25, 0.5, 0.75]),
- percentile_approx(col("foo"), (0.05, 0.95), 100),
- percentile_approx("foo", 0.5),
- percentile_approx("bar", [0.1, 0.9], lit(10)),
+ F.percentile_approx(F.col("foo"), F.lit(0.5)),
+ F.percentile_approx(F.col("bar"), 0.25, 42),
+ F.percentile_approx(F.col("bar"), [0.25, 0.5, 0.75]),
+ F.percentile_approx(F.col("foo"), (0.05, 0.95), 100),
+ F.percentile_approx("foo", 0.5),
+ F.percentile_approx("bar", [0.1, 0.9], F.lit(10)),
]
]
)
@@ -830,9 +731,6 @@ class FunctionsTestsMixin:
self.assertListEqual(actual, expected)
def test_nth_value(self):
- from pyspark.sql import Window
- from pyspark.sql.functions import nth_value
-
df = self.spark.createDataFrame(
[
("a", 0, None),
@@ -850,9 +748,9 @@ class FunctionsTestsMixin:
rs = df.select(
df.key,
df.order,
- nth_value("value", 2).over(w),
- nth_value("value", 2, False).over(w),
- nth_value("value", 2, True).over(w),
+ F.nth_value("value", 2).over(w),
+ F.nth_value("value", 2, False).over(w),
+ F.nth_value("value", 2, True).over(w),
).collect()
expected = [
@@ -869,11 +767,9 @@ class FunctionsTestsMixin:
self.assertEqual(tuple(r), ex[: len(r)])
def test_higher_order_function_failures(self):
- from pyspark.sql.functions import col, transform
-
# Should fail with varargs
with self.assertRaises(PySparkValueError) as pe:
- transform(col("foo"), lambda *x: lit(1))
+ F.transform(F.col("foo"), lambda *x: F.lit(1))
self.check_error(
exception=pe.exception,
@@ -883,7 +779,7 @@ class FunctionsTestsMixin:
# Should fail with kwargs
with self.assertRaises(PySparkValueError) as pe:
- transform(col("foo"), lambda **x: lit(1))
+ F.transform(F.col("foo"), lambda **x: F.lit(1))
self.check_error(
exception=pe.exception,
@@ -893,7 +789,7 @@ class FunctionsTestsMixin:
# Should fail with nullary function
with self.assertRaises(PySparkValueError) as pe:
- transform(col("foo"), lambda: lit(1))
+ F.transform(F.col("foo"), lambda: F.lit(1))
self.check_error(
exception=pe.exception,
@@ -903,7 +799,7 @@ class FunctionsTestsMixin:
# Should fail with quaternary function
with self.assertRaises(PySparkValueError) as pe:
- transform(col("foo"), lambda x1, x2, x3, x4: lit(1))
+ F.transform(F.col("foo"), lambda x1, x2, x3, x4: F.lit(1))
self.check_error(
exception=pe.exception,
@@ -913,7 +809,7 @@ class FunctionsTestsMixin:
# Should fail if function doesn't return Column
with self.assertRaises(PySparkValueError) as pe:
- transform(col("foo"), lambda x: 1)
+ F.transform(F.col("foo"), lambda x: 1)
self.check_error(
exception=pe.exception,
@@ -923,16 +819,14 @@ class FunctionsTestsMixin:
def test_nested_higher_order_function(self):
# SPARK-35382: lambda vars must be resolved properly in nested higher order functions
- from pyspark.sql.functions import flatten, struct, transform
-
df = self.spark.sql("SELECT array(1, 2, 3) as numbers, array('a', 'b', 'c') as letters")
actual = df.select(
- flatten(
- transform(
+ F.flatten(
+ F.transform(
"numbers",
- lambda number: transform(
- "letters", lambda letter: struct(number.alias("n"), letter.alias("l"))
+ lambda number: F.transform(
+ "letters", lambda letter: F.struct(number.alias("n"), letter.alias("l"))
),
)
)
@@ -955,7 +849,6 @@ class FunctionsTestsMixin:
def test_window_functions(self):
df = self.spark.createDataFrame([(1, "1"), (2, "2"), (1, "2"), (1, "2")], ["key", "value"])
w = Window.partitionBy("value").orderBy("key")
- from pyspark.sql import functions as F
sel = df.select(
df.value,
@@ -981,7 +874,6 @@ class FunctionsTestsMixin:
def test_window_functions_without_partitionBy(self):
df = self.spark.createDataFrame([(1, "1"), (2, "2"), (1, "2"), (1, "2")], ["key", "value"])
w = Window.orderBy("key", df.value)
- from pyspark.sql import functions as F
sel = df.select(
df.value,
@@ -1006,7 +898,6 @@ class FunctionsTestsMixin:
def test_window_functions_cumulative_sum(self):
df = self.spark.createDataFrame([("one", 1), ("two", 2)], ["key", "value"])
- from pyspark.sql import functions as F
# Test cumulative sum
sel = df.select(
@@ -1040,7 +931,6 @@ class FunctionsTestsMixin:
df = self.spark.createDataFrame(
[(datetime.datetime(2016, 3, 11, 9, 0, 7), 1)], ["date", "val"]
)
- from pyspark.sql import functions as F
w = df.groupBy(F.window("date", "5 seconds")).agg(F.sum("val").alias("sum"))
r = w.select(
@@ -1054,52 +944,43 @@ class FunctionsTestsMixin:
def test_collect_functions(self):
df = self.spark.createDataFrame([(1, "1"), (2, "2"), (1, "2"), (1, "2")], ["key", "value"])
- from pyspark.sql import functions
+ self.assertEqual(sorted(df.select(F.collect_set(df.key).alias("r")).collect()[0].r), [1, 2])
self.assertEqual(
- sorted(df.select(functions.collect_set(df.key).alias("r")).collect()[0].r), [1, 2]
- )
- self.assertEqual(
- sorted(df.select(functions.collect_list(df.key).alias("r")).collect()[0].r),
- [1, 1, 1, 2],
+ sorted(df.select(F.collect_list(df.key).alias("r")).collect()[0].r), [1, 1, 1, 2]
)
self.assertEqual(
- sorted(df.select(functions.collect_set(df.value).alias("r")).collect()[0].r), ["1", "2"]
+ sorted(df.select(F.collect_set(df.value).alias("r")).collect()[0].r), ["1", "2"]
)
self.assertEqual(
- sorted(df.select(functions.collect_list(df.value).alias("r")).collect()[0].r),
+ sorted(df.select(F.collect_list(df.value).alias("r")).collect()[0].r),
["1", "2", "2", "2"],
)
def test_datetime_functions(self):
- from pyspark.sql import functions
- from datetime import date
-
df = self.spark.range(1).selectExpr("'2017-01-22' as dateCol")
- parse_result = df.select(functions.to_date(functions.col("dateCol"))).first()
- self.assertEqual(date(2017, 1, 22), parse_result["to_date(dateCol)"])
+ parse_result = df.select(F.to_date(F.col("dateCol"))).first()
+ self.assertEqual(datetime.date(2017, 1, 22), parse_result["to_date(dateCol)"])
def test_assert_true(self):
self.check_assert_true(Py4JJavaError)
def check_assert_true(self, tpe):
- from pyspark.sql.functions import assert_true
-
df = self.spark.range(3)
self.assertEqual(
- df.select(assert_true(df.id < 3)).toDF("val").collect(),
+ df.select(F.assert_true(df.id < 3)).toDF("val").collect(),
[Row(val=None), Row(val=None), Row(val=None)],
)
with self.assertRaisesRegex(tpe, "too big"):
- df.select(assert_true(df.id < 2, "too big")).toDF("val").collect()
+ df.select(F.assert_true(df.id < 2, "too big")).toDF("val").collect()
with self.assertRaisesRegex(tpe, "2000000"):
- df.select(assert_true(df.id < 2, df.id * 1e6)).toDF("val").collect()
+ df.select(F.assert_true(df.id < 2, df.id * 1e6)).toDF("val").collect()
with self.assertRaises(PySparkTypeError) as pe:
- df.select(assert_true(df.id < 2, 5))
+ df.select(F.assert_true(df.id < 2, 5))
self.check_error(
exception=pe.exception,
@@ -1111,18 +992,16 @@ class FunctionsTestsMixin:
self.check_raise_error(Py4JJavaError)
def check_raise_error(self, tpe):
- from pyspark.sql.functions import raise_error
-
df = self.spark.createDataFrame([Row(id="foobar")])
with self.assertRaisesRegex(tpe, "foobar"):
- df.select(raise_error(df.id)).collect()
+ df.select(F.raise_error(df.id)).collect()
with self.assertRaisesRegex(tpe, "barfoo"):
- df.select(raise_error("barfoo")).collect()
+ df.select(F.raise_error("barfoo")).collect()
with self.assertRaises(PySparkTypeError) as pe:
- df.select(raise_error(None))
+ df.select(F.raise_error(None))
self.check_error(
exception=pe.exception,
@@ -1132,55 +1011,57 @@ class FunctionsTestsMixin:
def test_sum_distinct(self):
self.spark.range(10).select(
- assert_true(sum_distinct(col("id")) == sumDistinct(col("id")))
+ F.assert_true(F.sum_distinct(F.col("id")) == F.sumDistinct(F.col("id")))
).collect()
def test_shiftleft(self):
self.spark.range(10).select(
- assert_true(shiftLeft(col("id"), 2) == shiftleft(col("id"), 2))
+ F.assert_true(F.shiftLeft(F.col("id"), 2) == F.shiftleft(F.col("id"), 2))
).collect()
def test_shiftright(self):
self.spark.range(10).select(
- assert_true(shiftRight(col("id"), 2) == shiftright(col("id"), 2))
+ F.assert_true(F.shiftRight(F.col("id"), 2) == F.shiftright(F.col("id"), 2))
).collect()
def test_shiftrightunsigned(self):
self.spark.range(10).select(
- assert_true(shiftRightUnsigned(col("id"), 2) == shiftrightunsigned(col("id"), 2))
+ F.assert_true(
+ F.shiftRightUnsigned(F.col("id"), 2) == F.shiftrightunsigned(F.col("id"), 2)
+ )
).collect()
def test_lit_day_time_interval(self):
td = datetime.timedelta(days=1, hours=12, milliseconds=123)
- actual = self.spark.range(1).select(lit(td)).first()[0]
+ actual = self.spark.range(1).select(F.lit(td)).first()[0]
self.assertEqual(actual, td)
def test_lit_list(self):
# SPARK-40271: added list type supporting
test_list = [1, 2, 3]
expected = [1, 2, 3]
- actual = self.spark.range(1).select(lit(test_list)).first()[0]
+ actual = self.spark.range(1).select(F.lit(test_list)).first()[0]
self.assertEqual(actual, expected)
test_list = [[1, 2, 3], [3, 4]]
expected = [[1, 2, 3], [3, 4]]
- actual = self.spark.range(1).select(lit(test_list)).first()[0]
+ actual = self.spark.range(1).select(F.lit(test_list)).first()[0]
self.assertEqual(actual, expected)
with self.sql_conf({"spark.sql.ansi.enabled": False}):
test_list = ["a", 1, None, 1.0]
expected = ["a", "1", None, "1.0"]
- actual = self.spark.range(1).select(lit(test_list)).first()[0]
+ actual = self.spark.range(1).select(F.lit(test_list)).first()[0]
self.assertEqual(actual, expected)
test_list = [["a", 1, None, 1.0], [1, None, "b"]]
expected = [["a", "1", None, "1.0"], ["1", None, "b"]]
- actual = self.spark.range(1).select(lit(test_list)).first()[0]
+ actual = self.spark.range(1).select(F.lit(test_list)).first()[0]
self.assertEqual(actual, expected)
df = self.spark.range(10)
with self.assertRaises(PySparkValueError) as pe:
- lit([df.id, df.id])
+ F.lit([df.id, df.id])
self.check_error(
exception=pe.exception,
@@ -1196,8 +1077,8 @@ class FunctionsTestsMixin:
self.assertTrue(
all(
df.select(
- regexp_replace("str", r"(\d+)", "--") == "-----",
- regexp_replace("str", col("pattern"), col("replacement")) == "-----",
+ F.regexp_replace("str", r"(\d+)", "--") == "-----",
+ F.regexp_replace("str", F.col("pattern"), F.col("replacement")) == "-----",
).first()
)
)
@@ -1205,7 +1086,6 @@ class FunctionsTestsMixin:
@unittest.skipIf(not have_numpy, "NumPy not installed")
def test_lit_np_scalar(self):
import numpy as np
- from pyspark.sql.functions import lit
dtype_to_spark_dtypes = [
(np.int8, [("1", "tinyint")]),
@@ -1218,25 +1098,24 @@ class FunctionsTestsMixin:
]
for dtype, spark_dtypes in dtype_to_spark_dtypes:
with self.subTest(dtype):
- self.assertEqual(self.spark.range(1).select(lit(dtype(1))).dtypes, spark_dtypes)
+ self.assertEqual(self.spark.range(1).select(F.lit(dtype(1))).dtypes, spark_dtypes)
@unittest.skipIf(not have_numpy, "NumPy not installed")
def test_np_scalar_input(self):
import numpy as np
- from pyspark.sql.functions import array_contains, array_position
df = self.spark.createDataFrame([([1, 2, 3],), ([],)], ["data"])
for dtype in [np.int8, np.int16, np.int32, np.int64]:
- res = df.select(array_contains(df.data, dtype(1)).alias("b")).collect()
+ res = df.select(F.array_contains(df.data, dtype(1)).alias("b")).collect()
self.assertEqual([Row(b=True), Row(b=False)], res)
- res = df.select(array_position(df.data, dtype(1)).alias("c")).collect()
+ res = df.select(F.array_position(df.data, dtype(1)).alias("c")).collect()
self.assertEqual([Row(c=1), Row(c=0)], res)
df = self.spark.createDataFrame([([1.0, 2.0, 3.0],), ([],)], ["data"])
for dtype in [np.float32, np.float64]:
- res = df.select(array_contains(df.data, dtype(1)).alias("b")).collect()
+ res = df.select(F.array_contains(df.data, dtype(1)).alias("b")).collect()
self.assertEqual([Row(b=True), Row(b=False)], res)
- res = df.select(array_position(df.data, dtype(1)).alias("c")).collect()
+ res = df.select(F.array_position(df.data, dtype(1)).alias("c")).collect()
self.assertEqual([Row(c=1), Row(c=0)], res)
@unittest.skipIf(not have_numpy, "NumPy not installed")
@@ -1254,16 +1133,18 @@ class FunctionsTestsMixin:
for t, expected_spark_dtypes in arr_dtype_to_spark_dtypes:
arr = np.array([1, 2]).astype(t)
self.assertEqual(
- expected_spark_dtypes, self.spark.range(1).select(lit(arr).alias("b")).dtypes
+ expected_spark_dtypes, self.spark.range(1).select(F.lit(arr).alias("b")).dtypes
)
arr = np.array([1, 2]).astype(np.uint)
with self.assertRaisesRegex(
TypeError, "The type of array scalar '%s' is not supported" % arr.dtype
):
- self.spark.range(1).select(lit(arr).alias("b"))
+ self.spark.range(1).select(F.lit(arr).alias("b"))
def test_binary_math_function(self):
- funcs, expected = zip(*[(atan2, 0.13664), (hypot, 8.07527), (pow, 2.14359), (pmod, 1.1)])
+ funcs, expected = zip(
+ *[(F.atan2, 0.13664), (F.hypot, 8.07527), (F.pow, 2.14359), (F.pmod, 1.1)]
+ )
df = self.spark.range(1).select(*(func(1.1, 8) for func in funcs))
for a, e in zip(df.first(), expected):
self.assertAlmostEqual(a, e, 5)
@@ -1277,21 +1158,21 @@ class FunctionsTestsMixin:
)
actual = (
df.select(
- expr("map('c', 3, 'd', 4) as dict2"),
- map_from_arrays(df.k, df.v).alias("dict"),
+ F.expr("map('c', 3, 'd', 4) as dict2"),
+ F.map_from_arrays(df.k, df.v).alias("dict"),
"*",
)
.select(
- map_contains_key("dict", "a").alias("one"),
- map_contains_key("dict", "d").alias("not_exists"),
- map_keys("dict").alias("keys"),
- map_values("dict").alias("values"),
- map_entries("dict").alias("items"),
+ F.map_contains_key("dict", "a").alias("one"),
+ F.map_contains_key("dict", "d").alias("not_exists"),
+ F.map_keys("dict").alias("keys"),
+ F.map_values("dict").alias("values"),
+ F.map_entries("dict").alias("items"),
"*",
)
.select(
- map_concat("dict", "dict2").alias("merged"),
- map_from_entries(arrays_zip("keys", "values")).alias("from_items"),
+ F.map_concat("dict", "dict2").alias("merged"),
+ F.map_from_entries(F.arrays_zip("keys", "values")).alias("from_items"),
"*",
)
.first()
@@ -1307,7 +1188,7 @@ class FunctionsTestsMixin:
def test_schema_of_json(self):
with self.assertRaises(PySparkTypeError) as pe:
- schema_of_json(1)
+ F.schema_of_json(1)
self.check_error(
exception=pe.exception,
@@ -1317,7 +1198,7 @@ class FunctionsTestsMixin:
def test_schema_of_csv(self):
with self.assertRaises(PySparkTypeError) as pe:
- schema_of_csv(1)
+ F.schema_of_csv(1)
self.check_error(
exception=pe.exception,
@@ -1328,7 +1209,7 @@ class FunctionsTestsMixin:
def test_from_csv(self):
df = self.spark.range(10)
with self.assertRaises(PySparkTypeError) as pe:
- from_csv(df.id, 1)
+ F.from_csv(df.id, 1)
self.check_error(
exception=pe.exception,
@@ -1339,7 +1220,7 @@ class FunctionsTestsMixin:
def test_greatest(self):
df = self.spark.range(10)
with self.assertRaises(PySparkValueError) as pe:
- greatest(df.id)
+ F.greatest(df.id)
self.check_error(
exception=pe.exception,
@@ -1349,7 +1230,7 @@ class FunctionsTestsMixin:
def test_when(self):
with self.assertRaises(PySparkTypeError) as pe:
- when("id", 1)
+ F.when("id", 1)
self.check_error(
exception=pe.exception,
@@ -1359,7 +1240,7 @@ class FunctionsTestsMixin:
def test_window(self):
with self.assertRaises(PySparkTypeError) as pe:
- window("date", 5)
+ F.window("date", 5)
self.check_error(
exception=pe.exception,
@@ -1369,7 +1250,7 @@ class FunctionsTestsMixin:
def test_session_window(self):
with self.assertRaises(PySparkTypeError) as pe:
- session_window("date", 5)
+ F.session_window("date", 5)
self.check_error(
exception=pe.exception,
@@ -1379,7 +1260,7 @@ class FunctionsTestsMixin:
def test_bucket(self):
with self.assertRaises(PySparkTypeError) as pe:
- bucket("5", "id")
+ F.bucket("5", "id")
self.check_error(
exception=pe.exception,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org