python - read_parquet - dask vs spark

Aplicar la función al marco de datos agrupados en Dask: ¿Cómo se especifica el marco de datos agrupado como argumento en la función?

Question

Tengo un dask dataframe agrupado por el índice ( first_name ).

import pandas as pd
import numpy as np
from multiprocessing import cpu_count
from dask import dataframe as dd
from dask.multiprocessing import get
from dask.distributed import Client
NCORES = cpu_count()
client = Client()
entities = pd.DataFrame({''first_name'':[''Jake'',''John'',''Danae'',''Beatriz'', ''Jacke'', ''Jon''],''last_name'': [''Del Toro'', ''Foster'', ''Smith'', ''Patterson'', ''Toro'', ''Froster''], ''ID'':[''X'',''U'',''X'',''Y'', ''12'',''13'']})
df = dd.from_pandas(entities, npartitions=NCORES)
df = client.persist(df.set_index(''first_name''))

(Obviamente las entities en la vida real son varios miles de filas)

Quiero aplicar una función definida por el usuario a cada marco de datos agrupado. Quiero comparar cada fila con todas las demás filas del grupo (algo similar a Pandas compara cada fila con todas las filas en el marco de datos y guardar los resultados en la lista para cada fila ).

La siguiente es la función que trato de aplicar:

def contraster(x, DF):
matches = DF.apply(lambda row: fuzz.partial_ratio(row[''last_name''], x) >= 50, axis = 1)
return [i for i, x in enumerate(matches) if x]

Para el marco de datos de las entities prueba, podría aplicar la función como de costumbre:

entities.apply(lambda row: contraster(row[''last_name''], entities), axis =1)

Y el resultado esperado es:

Out[35]:
0 [0, 4]
1 [1, 5]
2 [2]
3 [3]
4 [0, 4]
5 [1, 5]
dtype: object

Cuando las entities son enormes, la solución es usar dask . Tenga en cuenta que el DF en la función de control debe ser el marco de datos agrupado.

Estoy tratando de usar lo siguiente:

df.groupby(''first_name'').apply(func=contraster, args=????)

Pero, ¿cómo debo especificar el marco de datos agrupado (es decir, DF en contraster ?)

Answer 1

Con un poco de conjeturas, creo que lo siguiente es lo que estás buscando.

def mapper(d):
def contraster(x, DF=d):
matches = DF.apply(lambda row: fuzz.partial_ratio(row[''last_name''], x) >= 50, axis = 1)
return [d.ID.iloc[i] for i, x in enumerate(matches) if x]
d[''out''] = d.apply(lambda row:
contraster(row[''last_name'']), axis =1)
return d
df.groupby(''first_name'').apply(mapper).compute()

Aplicado a tus datos, obtienes:

 ID first_name last_name out
2 X Danae Smith [X]
4 12 Jacke Toro [12]
0 X Jake Del Toro [X]
1 U John Foster [U]
5 13 Jon Froster [13]
3 Y Beatriz Patterson [Y]

es decir, debido a que se agrupa por el primer nombre, cada grupo solo contiene un elemento, que coincide solo con sí mismo.

Sin embargo, si tiene algunos valores de primer nombre que estaban en varias filas, obtendría coincidencias:

entities = pd.DataFrame(
{''first_name'':[''Jake'',''Jake'', ''Jake'', ''John''],
''last_name'': [''Del Toro'', ''Toro'', ''Smith''
''Froster''],
''ID'':[''Z'',''U'',''X'',''Y'']})

Salida:

 ID first_name last_name out
0 Z Jake Del Toro [Z, U]
1 U Jake Toro [Z, U]
2 X Jake Smith [X]
3 Y John Froster [Y]

Si no necesita coincidencias exactas en el primer nombre, entonces tal vez necesite ordenar / establecer el índice por el primer nombre y usar map_partitions de una manera similar. En ese caso, necesitarás reformar tu pregunta.

Answer 2

La función que proporciona a groupby-apply debe tomar un marco de datos o una serie de Pandas como entrada y, idealmente, devolver uno (o un valor escalar) como salida. Los parámetros adicionales están bien, pero deberían ser secundarios, no el primer argumento. Esto es lo mismo en los marcos de datos de Pandas y Dask.

def func(df, x=None):
# do whatever you want here
# the input to this function will have all the same first name
return pd.DataFrame({''x'': [x] * len(df),
''count'': len(df),
''first_name'': df.first_name})

A continuación, puede llamar a df.groupby normalmente

import pandas as pd
import dask.dataframe as dd
df = pd.DataFrame({''first_name'':[''Alice'', ''Alice'', ''Bob''],
''last_name'': [''Adams'', ''Jones'', ''Smith'']})
ddf = dd.from_pandas(df, npartitions=2)
ddf.groupby(''first_name'').apply(func, x=3).compute()

Esto producirá la misma salida en pandas o dask.dataframe

 count first_name x
0 2 Alice 3
1 2 Alice 3
2 1 Bob 3