# # Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # import datetime import unittest from distutils.version import LooseVersion import numpy as np import pandas as pd from pyspark import pandas as ps from pyspark.testing.pandasutils import PandasOnSparkTestCase from pyspark.testing.sqlutils import SQLTestUtils class SeriesDateTimeTestsMixin: @property def pdf1(self): date1 = pd.Series(pd.date_range("2012-1-1 12:45:31", periods=3, freq="M")) date2 = pd.Series(pd.date_range("2013-3-11 21:45:00", periods=3, freq="W")) return pd.DataFrame(dict(start_date=date1, end_date=date2)) @property def pd_start_date(self): return self.pdf1["start_date"] @property def ks_start_date(self): return ps.from_pandas(self.pd_start_date) def check_func(self, func): self.assert_eq(func(self.ks_start_date), func(self.pd_start_date)) @unittest.skip( "Those fail in certain OSs presumably due to different" "timezone behaviours inherited from C library." ) def test_timestamp_subtraction(self): pdf = self.pdf1 psdf = ps.from_pandas(pdf) actual = (psdf["end_date"] - psdf["start_date"] - 1)._to_pandas() expected = (pdf["end_date"] - pdf["start_date"]) // np.timedelta64(1, "s") - 1 self.assert_eq(actual, expected) actual = (psdf["end_date"] - pd.Timestamp("2012-1-1 12:45:31") - 1)._to_pandas() expected = (pdf["end_date"] - pd.Timestamp("2012-1-1 12:45:31")) // np.timedelta64( 1, "s" ) - 1 self.assert_eq(actual, expected) actual = (pd.Timestamp("2013-3-11 21:45:00") - psdf["start_date"] - 1)._to_pandas() expected = (pd.Timestamp("2013-3-11 21:45:00") - pdf["start_date"]) // np.timedelta64( 1, "s" ) - 1 self.assert_eq(actual, expected) def test_timestamp_subtraction_errors(self): psdf = ps.DataFrame( {"a": pd.date_range("2016-12-31", "2017-01-08", freq="D"), "b": pd.Series(range(9))} ) expected_error_message = "Datetime subtraction can only be applied to datetime series." with self.assertRaisesRegex(TypeError, expected_error_message): psdf["a"] - psdf["b"] with self.assertRaisesRegex(TypeError, expected_error_message): psdf["a"] - 1 with self.assertRaisesRegex(TypeError, expected_error_message): 1 - psdf["a"] def test_arithmetic_op_exceptions(self): psser = self.ks_start_date py_datetime = self.pd_start_date.dt.to_pydatetime() datetime_index = ps.Index(self.pd_start_date) for other in [1, 0.1, psser, datetime_index, py_datetime]: expected_err_msg = "Addition can not be applied to datetimes." self.assertRaisesRegex(TypeError, expected_err_msg, lambda: psser + other) self.assertRaisesRegex(TypeError, expected_err_msg, lambda: other + psser) expected_err_msg = "Multiplication can not be applied to datetimes." self.assertRaisesRegex(TypeError, expected_err_msg, lambda: psser * other) self.assertRaisesRegex(TypeError, expected_err_msg, lambda: other * psser) expected_err_msg = "True division can not be applied to datetimes." self.assertRaisesRegex(TypeError, expected_err_msg, lambda: psser / other) self.assertRaisesRegex(TypeError, expected_err_msg, lambda: other / psser) expected_err_msg = "Floor division can not be applied to datetimes." self.assertRaisesRegex(TypeError, expected_err_msg, lambda: psser // other) self.assertRaisesRegex(TypeError, expected_err_msg, lambda: other // psser) expected_err_msg = "Modulo can not be applied to datetimes." self.assertRaisesRegex(TypeError, expected_err_msg, lambda: psser % other) self.assertRaisesRegex(TypeError, expected_err_msg, lambda: other % psser) expected_err_msg = "Datetime subtraction can only be applied to datetime series." for other in [1, 0.1]: self.assertRaisesRegex(TypeError, expected_err_msg, lambda: psser - other) self.assertRaisesRegex(TypeError, expected_err_msg, lambda: other - psser) self.assertRaisesRegex(TypeError, expected_err_msg, lambda: psser - other) self.assertRaises(NotImplementedError, lambda: py_datetime - psser) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43462): Enable SeriesDateTimeTests.test_date_subtraction for pandas 2.0.0.", ) def test_date_subtraction(self): pdf = self.pdf1 psdf = ps.from_pandas(pdf) self.assert_eq( psdf["end_date"].dt.date - psdf["start_date"].dt.date, (pdf["end_date"].dt.date - pdf["start_date"].dt.date).dt.days, ) self.assert_eq( psdf["end_date"].dt.date - datetime.date(2012, 1, 1), (pdf["end_date"].dt.date - datetime.date(2012, 1, 1)).dt.days, ) self.assert_eq( datetime.date(2013, 3, 11) - psdf["start_date"].dt.date, (datetime.date(2013, 3, 11) - pdf["start_date"].dt.date).dt.days, ) psdf = ps.DataFrame( {"a": pd.date_range("2016-12-31", "2017-01-08", freq="D"), "b": pd.Series(range(9))} ) expected_error_message = "Date subtraction can only be applied to date series." with self.assertRaisesRegex(TypeError, expected_error_message): psdf["a"].dt.date - psdf["b"] with self.assertRaisesRegex(TypeError, expected_error_message): psdf["a"].dt.date - 1 with self.assertRaisesRegex(TypeError, expected_error_message): 1 - psdf["a"].dt.date @unittest.skip( "It fails in certain OSs presumably due to different " "timezone behaviours inherited from C library." ) def test_div(self): pdf = self.pdf1 psdf = ps.from_pandas(pdf) for u in "D", "s", "ms": duration = np.timedelta64(1, u) self.assert_eq( (psdf["end_date"] - psdf["start_date"]) / duration, (pdf["end_date"] - pdf["start_date"]) / duration, ) @unittest.skip("It is currently failed probably for the same reason in 'test_subtraction'") def test_date(self): self.check_func(lambda x: x.dt.date) def test_time(self): with self.assertRaises(NotImplementedError): self.check_func(lambda x: x.dt.time) def test_timetz(self): with self.assertRaises(NotImplementedError): self.check_func(lambda x: x.dt.timetz) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43736): Enable SeriesDateTimeTests.test_year for pandas 2.0.0.", ) def test_year(self): self.check_func(lambda x: x.dt.year) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43731): Enable SeriesDateTimeTests.test_month for pandas 2.0.0.", ) def test_month(self): self.check_func(lambda x: x.dt.month) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43722): Enable SeriesDateTimeTests.test_day for pandas 2.0.0.", ) def test_day(self): self.check_func(lambda x: x.dt.day) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43728): Enable SeriesDateTimeTests.test_hour for pandas 2.0.0.", ) def test_hour(self): self.check_func(lambda x: x.dt.hour) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43730): Enable SeriesDateTimeTests.test_minute for pandas 2.0.0.", ) def test_minute(self): self.check_func(lambda x: x.dt.minute) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43733): Enable SeriesDateTimeTests.test_second for pandas 2.0.0.", ) def test_second(self): self.check_func(lambda x: x.dt.second) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43729): Enable SeriesDateTimeTests.test_microsecond for pandas 2.0.0.", ) def test_microsecond(self): self.check_func(lambda x: x.dt.microsecond) def test_nanosecond(self): with self.assertRaises(NotImplementedError): self.check_func(lambda x: x.dt.nanosecond) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-42617): Support `isocalendar`", ) def test_week(self): self.check_func(lambda x: x.dt.week) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-42617): Support `isocalendar`", ) def test_weekofyear(self): self.check_func(lambda x: x.dt.weekofyear) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43723): Enable SeriesDateTimeTests.test_dayofweek for pandas 2.0.0.", ) def test_dayofweek(self): self.check_func(lambda x: x.dt.dayofweek) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43735): Enable SeriesDateTimeTests.test_weekday for pandas 2.0.0.", ) def test_weekday(self): self.check_func(lambda x: x.dt.weekday) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43724): Enable SeriesDateTimeTests.test_dayofyear for pandas 2.0.0.", ) def test_dayofyear(self): self.check_func(lambda x: x.dt.dayofyear) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43732): Enable SeriesDateTimeTests.test_quarter for pandas 2.0.0.", ) def test_quarter(self): self.check_func(lambda x: x.dt.quarter) def test_is_month_start(self): self.check_func(lambda x: x.dt.is_month_start) def test_is_month_end(self): self.check_func(lambda x: x.dt.is_month_end) def test_is_quarter_start(self): self.check_func(lambda x: x.dt.is_quarter_start) def test_is_quarter_end(self): self.check_func(lambda x: x.dt.is_quarter_end) def test_is_year_start(self): self.check_func(lambda x: x.dt.is_year_start) def test_is_year_end(self): self.check_func(lambda x: x.dt.is_year_end) def test_is_leap_year(self): self.check_func(lambda x: x.dt.is_leap_year) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43726): Enable SeriesDateTimeTests.test_daysinmonth for pandas 2.0.0.", ) def test_daysinmonth(self): self.check_func(lambda x: x.dt.daysinmonth) @unittest.skipIf( LooseVersion(pd.__version__) >= LooseVersion("2.0.0"), "TODO(SPARK-43725): Enable SeriesDateTimeTests.test_days_in_month for pandas 2.0.0.", ) def test_days_in_month(self): self.check_func(lambda x: x.dt.days_in_month) @unittest.expectedFailure def test_tz_localize(self): self.check_func(lambda x: x.dt.tz_localize("America/New_York")) @unittest.expectedFailure def test_tz_convert(self): self.check_func(lambda x: x.dt.tz_convert("America/New_York")) def test_normalize(self): self.check_func(lambda x: x.dt.normalize()) def test_strftime(self): self.check_func(lambda x: x.dt.strftime("%Y-%m-%d")) def test_round(self): self.check_func(lambda x: x.dt.round(freq="min")) self.check_func(lambda x: x.dt.round(freq="H")) def test_floor(self): self.check_func(lambda x: x.dt.floor(freq="min")) self.check_func(lambda x: x.dt.floor(freq="H")) def test_ceil(self): self.check_func(lambda x: x.dt.ceil(freq="min")) self.check_func(lambda x: x.dt.ceil(freq="H")) @unittest.skip("Unsupported locale setting") def test_month_name(self): self.check_func(lambda x: x.dt.month_name()) self.check_func(lambda x: x.dt.month_name(locale="en_US.UTF-8")) @unittest.skip("Unsupported locale setting") def test_day_name(self): self.check_func(lambda x: x.dt.day_name()) self.check_func(lambda x: x.dt.day_name(locale="en_US.UTF-8")) def test_unsupported_type(self): self.assertRaisesRegex( ValueError, "Cannot call DatetimeMethods on type LongType", lambda: ps.Series([0]).dt ) class SeriesDateTimeTests(SeriesDateTimeTestsMixin, PandasOnSparkTestCase, SQLTestUtils): pass if __name__ == "__main__": from pyspark.pandas.tests.test_series_datetime import * # noqa: F401 try: import xmlrunner testRunner = xmlrunner.XMLTestRunner(output="target/test-reports", verbosity=2) except ImportError: testRunner = None unittest.main(testRunner=testRunner, verbosity=2)