code: def test_groupby_aggregate_with_start_state(stream):
      example = cudf.DataFrame({"name": [], "amount": []})
      sdf = DataFrame(stream, example=example).groupby(["name"])
      output0 = (
          sdf.amount.sum(start=None).stream.gather().sink_to_list()
      )
      output1 = (
          sdf.amount.mean(with_state=True, start=None)
          .stream.gather()
          .sink_to_list()
      )
      output2 = (
          sdf.amount.count(start=None).stream.gather().sink_to_list()
      )
  
      df = cudf.DataFrame(
          {"name": ["Alice", "Tom"], "amount": [50, 100]}
      )
      stream.emit(df)
  
      out_df0 = cudf.DataFrame(
          {"name": ["Alice", "Tom"], "amount": [50, 100]}
      )
      out_df1 = cudf.DataFrame(
          {"name": ["Alice", "Tom"], "amount": [50.0, 100.0]}
      )
      out_df2 = cudf.DataFrame(
          {"name": ["Alice", "Tom"], "amount": [1, 1]}
      )
      assert assert_eq(output0[0].reset_index(), out_df0)
      assert assert_eq(output1[0][1].reset_index(), out_df1)
      assert assert_eq(output2[0].reset_index(), out_df2)
  
      example = cudf.DataFrame({"name": [], "amount": []})
      sdf = DataFrame(stream, example=example).groupby(["name"])
      output3 = (
          sdf.amount.sum(start=output0[0])
          .stream.gather()
          .sink_to_list()
      )
      output4 = (
          sdf.amount.mean(with_state=True, start=output1[0][0])
          .stream.gather()
          .sink_to_list()
      )
      output5 = (
          sdf.amount.count(start=output2[0])
          .stream.gather()
          .sink_to_list()
      )
      df = cudf.DataFrame(
          {"name": ["Alice", "Tom", "Linda"], "amount": [50, 100, 200]}
      )
      stream.emit(df)
  
      out_df2 = cudf.DataFrame(
          {"name": ["Alice", "Linda", "Tom"], "amount": [100, 200, 200]}
      )
      out_df3 = cudf.DataFrame(
          {
              "name": ["Alice", "Linda", "Tom"],
              "amount": [50.0, 200.0, 100.0],
          }
      )
      out_df4 = cudf.DataFrame(
          {"name": ["Alice", "Linda", "Tom"], "amount": [2, 1, 2]}
      )
      assert assert_eq(output3[0].reset_index(), out_df2)
      assert assert_eq(output4[0][1].reset_index(), out_df3)
      assert assert_eq(output5[0].reset_index(), out_df4)
  
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 38 for calling a global function with example usage: test_groupby_aggregate_with_start_state(stream) without return types