Aggregating data every n elements

Abierto
#351 3 comentarios 0 reacciones 0 asignados Ver en GitHub

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
5/5
Tiempo estimado
Más de una semana
Aptitud para principiantes
25/100
Tipo de issue
Nueva funcionalidad
Claridad
Necesita aclaración
Estado de actividad
Estancado
Stack tecnológico
pandas, python
Área
data, stream-processing

Línea de trabajo

Start with the reported stream.map(...).partition(100).to_batch().map_partitions(func) chain and inspect the map_partitions entry point, especially how its example is inferred. Determine whether chunked pandas value_counts aggregation is supported or requires a new API, then define a reproducible example and tests that establish the expected output for successive 100-element batches.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

Hi,

I would like to calculate the .value_counts() from a pandas dataframe in chunks of n elements and output the aggregated result.
For example, I have 1000 elements. Calculate the value_counts() for the first 100 and output the result, aggregate the result to the next 100 elements and output it again.

I tried

def func(df):
    return df
stream.map(df["OFFENSE_CODE"]).partition(100).to_batch().map_partitions(func)

but I get

TypeError                                 Traceback (most recent call last)
<ipython-input-291-e68a39a9e8bb> in <module>
      1 def func(df):
      2     return df
----> 3 stream.map(df["OFFENSE_CODE"]).partition(100).to_batch().map_partitions(func)
      4 
      5 # a.value_counts()

~\Anaconda3\lib\site-packages\streamz\collection.py in map_partitions(func, *args, **kwargs)
     19     example = kwargs.pop('example', None)
     20     if example is None:
---> 21         example = func(*[getattr(arg, 'example', arg) for arg in args], **kwargs)
     22 
     23     streams = [arg for arg in args if isinstance(arg, Streaming)]

TypeError: func() missing 1 required positional argument: 'df'

BTW I am not sure this is the best approach to do this. Any hint?

Lenguaje dominante
Python
Estrellas
1.3k
Forks
149
Sin PR fusionados en 30 d

Guía de contribución

Abrir la guía de contribución

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de python-streamz/streamz

Todos los issues de python-streamz/streamz

Issues similares

Más issues de Python

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.