code: def test_grouped_over_window_with_key(self):
      data = [
          (0, 1, "2018-03-10T00:00:00+00:00", [0]),
          (1, 2, "2018-03-11T00:00:00+00:00", [0]),
          (2, 2, "2018-03-12T00:00:00+00:00", [0]),
          (3, 3, "2018-03-15T00:00:00+00:00", [0]),
          (4, 3, "2018-03-16T00:00:00+00:00", [0]),
          (5, 3, "2018-03-17T00:00:00+00:00", [0]),
          (6, 3, "2018-03-21T00:00:00+00:00", [0]),
      ]
  
      timezone = self.spark.conf.get("spark.sql.session.timeZone")
      expected_window = [
          {
              key: (
                  pd.Timestamp(ts)
                  .tz_localize(datetime.timezone.utc)
                  .tz_convert(timezone)
                  .tz_localize(None)
              )
              for key, ts in w.items()
          }
          for w in [
              {
                  "start": datetime.datetime(2018, 3, 10, 0, 0),
                  "end": datetime.datetime(2018, 3, 15, 0, 0),
              },
              {
                  "start": datetime.datetime(2018, 3, 15, 0, 0),
                  "end": datetime.datetime(2018, 3, 20, 0, 0),
              },
              {
                  "start": datetime.datetime(2018, 3, 20, 0, 0),
                  "end": datetime.datetime(2018, 3, 25, 0, 0),
              },
          ]
      ]
  
      expected_key = {
          0: (1, expected_window[0]),
          1: (2, expected_window[0]),
          2: (2, expected_window[0]),
          3: (3, expected_window[1]),
          4: (3, expected_window[1]),
          5: (3, expected_window[1]),
          6: (3, expected_window[2]),
      }
  
      # id -> array of group with len of num records in window
      expected = {
          0: [1],
          1: [2, 2],
          2: [2, 2],
          3: [3, 3, 3],
          4: [3, 3, 3],
          5: [3, 3, 3],
          6: [3],
      }
  
      df = self.spark.createDataFrame(
          data, ["id", "group", "ts", "result"]
      )
      df = df.select(
          col("id"),
          col("group"),
          col("ts").cast("timestamp"),
          col("result"),
      )
  
      def f(key, pdf):
          group = key[0]
          window_range = key[1]
  
          # Make sure the key with group and window values are correct
          for _, i in pdf.id.items():
              assert expected_key[i][0] == group, "{} != {}".format(
                  expected_key[i][0], group
              )
              assert (
                  expected_key[i][1] == window_range
              ), "{} != {}".format(expected_key[i][1], window_range)
  
          return pdf.assign(result=[[group] * len(pdf)] * len(pdf))
  
      result = (
          df.groupby("group", window("ts", "5 days"))
          .applyInPandas(f, df.schema)
          .select("id", "result")
          .orderBy("id")
          .collect()
      )
  
      self.assertListEqual(
          [Row(id=key, result=val) for key, val in expected.items()],
          result,
      )
  
imports: import datetime
  import unittest
  from collections import OrderedDict
  from decimal import Decimal
  from typing import cast
  from pyspark.sql import Row
  from pyspark.sql.functions import array, explode, col, lit, udf, sum, pandas_udf, PandasUDFType, window
  from pyspark.sql.types import IntegerType, DoubleType, ArrayType, BinaryType, ByteType, LongType, DecimalType, ShortType, FloatType, StringType, BooleanType, StructType, StructField, NullType, MapType, YearMonthIntervalType
  from pyspark.errors import PythonException, PySparkTypeError
  from pyspark.testing.sqlutils import ReusedSQLTestCase, have_pandas, have_pyarrow, pandas_requirement_message, pyarrow_requirement_message
  from pyspark.testing.utils import QuietTest
label_desc: Use image node_id 40 for calling the GroupedApplyInPandasTestsMixin obj's underlying member method code with example usage: obj.test_grouped_over_window_with_key() and returns: pdf