code: def test_groupby_aggregate(agg, grouper, indexer, stream):
      df = cudf.DataFrame(
          {"x": (np.arange(10) // 2).astype(float), "y": [1.0, 2.0] * 5}
      )
  
      a = DataFrame(example=df.iloc[:0], stream=stream)
  
      def f(x):
          return agg(indexer(x.groupby(grouper(x))))
  
      L = f(a).stream.gather().sink_to_list()
  
      a.emit(df.iloc[:3])
      a.emit(df.iloc[3:7])
      a.emit(df.iloc[7:])
  
      first = df.iloc[:3]
      g = f(first)
  
      h = f(df)
  
      assert_eq(L[0], g)
      assert_eq(L[-1], h)
  
imports: import json
  import operator
  import numpy
  import pandas
  import pytest
  from dask.dataframe.utils import assert_eq
  from distributed import Client
  from streamz import Stream
  from streamz.dask import DaskStream
  from streamz.dataframe import Aggregation, DataFrame, DataFrames, Series
label_desc: Use image node_id 16 for calling a global function with example usage: test_groupby_aggregate(agg, grouper, indexer, stream) and returns: agg