code: def test_complex_group_by(self):
      left = pd.DataFrame.from_dict(
          {"id": [1, 2, 3], "k": [5, 6, 7], "v": [9, 10, 11]}
      )
  
      right = pd.DataFrame.from_dict(
          {"id": [11, 12, 13], "k": [5, 6, 7], "v2": [90, 100, 110]}
      )
  
      left_gdf = self.spark.createDataFrame(left).groupby(
          col("id") % 2 == 0
      )
  
      right_gdf = self.spark.createDataFrame(right).groupby(
          col("id") % 2 == 0
      )
  
      def merge_pandas(lft, rgt):
          return pd.merge(lft[["k", "v"]], rgt[["k", "v2"]], on=["k"])
  
      result = (
          left_gdf.cogroup(right_gdf)
          .applyInPandas(merge_pandas, "k long, v long, v2 long")
          .sort(["k"])
          .toPandas()
      )
  
      expected = pd.DataFrame.from_dict(
          {"k": [5, 6, 7], "v": [9, 10, 11], "v2": [90, 100, 110]}
      )
  
      assert_frame_equal(expected, result)
  
imports: import unittest
  from typing import cast
  from pyspark.sql.functions import array, explode, col, lit, udf, pandas_udf, sum
  from pyspark.sql.types import ArrayType, DoubleType, LongType, StructType, StructField, YearMonthIntervalType, Row
  from pyspark.sql.window import Window
  from pyspark.errors import IllegalArgumentException, PythonException
  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 8 for calling the CogroupedApplyInPandasTestsMixin obj's underlying member method code with example usage: obj.test_complex_group_by() and returns: pd