code: def test_integration_from_stream(stream):
      source = stream
      sdf = (
          source.partition(4)
          .to_batch(example=['{"x": 0, "y": 0}'])
          .map(json.loads)
          .to_dataframe()
      )
      result = sdf.groupby(sdf.x).y.sum().mean()
      L = result.stream.gather().sink_to_list()
  
      for i in range(12):
          source.emit(json.dumps({"x": i % 3, "y": i}))
  
      assert L == [2, 28 / 3, 22.0]
  
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 23 for calling a global function with example usage: test_integration_from_stream(stream) without return types