code: def test_custom_aggregation():
      df = cudf.DataFrame(
          {"x": np.arange(10, dtype=float), "y": [1.0, 2.0] * 5}
      )
  
      class Custom(Aggregation):
          def initial(self, new):
              return 0
  
          def on_new(self, state, new):
              return state + 1, state
  
          def on_old(self, state, new):
              return state - 100, state
  
      sdf = DataFrame(example=df)
      L = sdf.aggregate(Custom()).stream.sink_to_list()
  
      sdf.emit(df)
      sdf.emit(df)
      sdf.emit(df)
  
      assert L == [0, 1, 2]
  
      sdf = DataFrame(example=df)
      L = sdf.window(n=5).aggregate(Custom()).stream.sink_to_list()
  
      sdf.emit(df)
      sdf.emit(df)
      sdf.emit(df)
  
      assert L == [1, -198, -397]
  
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 37 for calling a global function with example usage: test_custom_aggregation() and returns: int, unknown,  state, unknown,  state