La API de MapReduce de PyODPS DataFrame permite escribir lógica personalizada de map y reduce en Python para ejecutarla a gran escala en MaxCompute. Un trabajo de map_reduce puede contener solo mappers, solo reducers o ambos.
Ejemplo de WordCount
El siguiente ejemplo cuenta las apariciones de palabras en una tabla que contiene una sola columna STRING.
#encoding=utf-8
from odps import ODPS
from odps import options
from odps.df import DataFrame
options.verbose = True
o = ODPS('your-access-id', 'your-secret-access-key',
project='DMP_UC_dev',
endpoint='http://service-corp.odps.aliyun-inc.com/api')
def mapper(row):
for word in row[0].split():
yield word.lower(), 1
def reducer(keys):
# Use a list instead of cnt=0. A plain integer would be treated as a local
# variable inside h(), so its value would not appear in the output.
cnt = [0]
def h(row, done): # done=True when all rows for this key have been processed
cnt[0] += row[1]
if done:
yield keys[0], cnt[0]
return h
word_count = DataFrame(o.get_table('zx_word_count'))
table = word_count.map_reduce(
mapper, reducer,
group=['word'],
mapper_output_names=['word', 'cnt'],
mapper_output_types=['string', 'int'],
reducer_output_names=['word', 'cnt'],
reducer_output_types=['string', 'int'],
)Salida esperada:
word cnt
0 are 1
1 day 1
2 doing? 1
...El parámetro group especifica el campo que utiliza el reducer para agrupar las filas entrantes. Si se omite, se usan todos los campos para la agrupación. El reducer recibe las keys agregadas y procesa cada fila que comparte dichas claves. La bandera done es True cuando se ha procesado la última fila correspondiente a una clave determinada.
Es posible escribir el reducer como una clase invocable en lugar de usar un cierre:
class reducer(object):
def __init__(self, keys):
self.cnt = 0
def __call__(self, row, done): # done=True when all rows for this key are processed
self.cnt += row.cnt
if done:
yield row.word, self.cntSimplificación del esquema de salida con el decorador @output
Utilice el decorador @output para declarar los nombres y tipos de los campos de salida directamente en la función. Esto elimina la necesidad de pasar mapper_output_names, mapper_output_types, reducer_output_names y reducer_output_types a map_reduce.
from odps.df import output
@output(['word', 'cnt'], ['string', 'int'])
def mapper(row):
for word in row[0].split():
yield word.lower(), 1
@output(['word', 'cnt'], ['string', 'int'])
def reducer(keys):
cnt = [0]
def h(row, done):
cnt[0] += row.cnt
if done:
yield keys.word, cnt[0]
return h
word_count = DataFrame(o.get_table('zx_word_count'))
table = word_count.map_reduce(mapper, reducer, group='word')Para ordenar las filas dentro de cada grupo durante la iteración, pase el parámetro sort. El parámetro ascending controla la dirección de la ordenación: un solo valor booleano aplica el mismo orden a todos los campos de sort; una lista permite establecer una dirección diferente por campo (la longitud de la lista debe coincidir con el número de campos de sort).
Especificación de un combiner
Durante un trabajo de MapReduce, la salida del mapper se transfiere a través de la red hacia los reducers, en un paso denominado shuffle. Este proceso implica E/S de disco, serialización de datos y transferencia de red, por lo que constituye una de las partes más costosas del pipeline.
Un combiner reduce el costo del shuffle al agregar localmente la salida del mapper en cada nodo antes de enviarla a los reducers. Esta estrategia resulta más eficaz para operaciones conmutativas y asociativas, como conteos y sumas.
Restricciones previas a la escritura de un combiner:
Un combiner no puede hacer referencia a recursos.
Los nombres y tipos de los campos de salida deben coincidir exactamente con los del mapper correspondiente.
La interfaz del combiner es idéntica a la del reducer. Pase la función del combiner mediante el parámetro combiner:
words_df.map_reduce(mapper, reducer, combiner=reducer, group='word')Referencia a recursos
Tanto los mappers como los reducers pueden referenciar su propio conjunto de recursos. Los recursos se pasan como parámetros de la función (por ejemplo, def mapper(resources):).
El siguiente ejemplo filtra palabras vacías en el mapper e incrementa en 5 el conteo de las palabras incluidas en una lista blanca dentro del reducer:
white_list_file = o.create_resource('pyodps_white_list_words', 'file', file_obj='Python\nWorld')
@output(['word', 'cnt'], ['string', 'int'])
def mapper(resources):
stop_words = set(r[0].strip() for r in resources[0])
def h(row):
for word in row[0].split():
if word not in stop_words:
yield word, 1
return h
@output(['word', 'cnt'], ['string', 'int'])
def reducer(resources):
d = dict()
d['white_list'] = set(word.strip() for word in resources[0])
d['cnt'] = 0
def inner(keys):
d['cnt'] = 0
def h(row, done):
d['cnt'] += row.cnt
if done:
if row.word in d['white_list']:
d['cnt'] += 5
yield keys.word, d['cnt']
return h
return inner
words_df.map_reduce(mapper, reducer, group='word',
mapper_resources=[stop_words],
reducer_resources=[white_list_file])Salida esperada:
word cnt
0 hello 2
1 life 1
2 python 7
3 world 6
4 short 1
5 use 1Uso de bibliotecas de Python de terceros
Las características de bytecode de Python 3, como yield from, provocan errores en los workers de MaxCompute que ejecutan Python 2.7. Pruebe su código de extremo a extremo antes de ejecutar trabajos de MapReduce basados en Python 3 en producción.
Especifique las bibliotecas de forma global para aplicarlas a todas las llamadas de map_reduce en una sesión:
from odps import options
options.df.libraries = ['six.whl', 'python_dateutil.whl']O bien páselas a una ejecución individual:
df.map_reduce(mapper=my_mapper, reducer=my_reducer, group='key').execute(
libraries=['six.whl', 'python_dateutil.whl']
)Reorganización de datos
Cuando los datos se distribuyen de manera desigual entre las particiones del clúster, llame a reshuffle para reequilibrarlos. De forma predeterminada, las filas se asignan a las particiones mediante un hash aleatorio:
df1 = df.reshuffle()Para distribuir por una columna específica y ordenar el resultado:
df1 = df.reshuffle('name', sort='id', ascending=False)Filtro Bloom
bloom_filter prefiltra rápidamente un conjunto de datos frente a otro antes de realizar una unión, lo que reduce la cantidad de filas que requieren shuffle y comparación. Funciona mejor cuando un conjunto de datos es mucho mayor que el otro; por ejemplo, al filtrar datos de eventos de navegación contra un conjunto menor de registros de transacciones antes de unirlos.
El filtro es aproximado: elimina las filas que definitivamente están ausentes del conjunto de referencia, pero puede conservar un pequeño número de filas que en realidad no están presentes.
df1 = DataFrame(pd.DataFrame({'a': ['name1', 'name2', 'name3', 'name1'], 'b': [1, 2, 3, 4]}))
df2 = DataFrame(pd.DataFrame({'a': ['name1']}))
df1.bloom_filter('a', df2.a)
# The first argument can also be a column expression, e.g., df1.a + '1'Salida esperada:
a b
0 name1 1
1 name1 4En este ejemplo, name2 y name3 se filtran. En conjuntos de datos más grandes, es posible que el filtro no elimine todas las filas sin coincidencia, pero el resultado de la unión sigue siendo correcto, ya que el filtro solo afecta el rendimiento y no la precisión.
Configure el filtro con estos parámetros:
|
Parameter |
Default |
Effect on accuracy |
Effect on memory |
|
|
|
Valores más altos reducen los falsos positivos |
Aumenta |
|
|
|
Valores más bajos reducen los falsos positivos |
Aumenta |
Establezca ambos parámetros según el tamaño de su conjunto de datos y la memoria disponible.
Para obtener información sobre la ejecución de operaciones de DataFrame, consulte DataFrame execution.
Tabla dinámica
pivot_table resume los datos mediante la agrupación de filas y el cálculo de valores agregados en las columnas.
Datos de muestra:
>>> df
A B C D E
0 foo one small 1 3
1 foo one large 2 4
2 foo one large 2 5
3 foo two small 3 6
4 foo two small 3 4
5 bar one large 4 5
6 bar one small 5 3
7 bar two small 6 2
8 bar two large 7 1El parámetro rows es obligatorio. Especifica los campos por los cuales agrupar; la función de agregación predeterminada es mean:
>>> df['A', 'D', 'E'].pivot_table(rows='A')
A D_mean E_mean
0 bar 5.5 2.75
1 foo 2.2 4.40Pase varios campos a rows para lograr una agrupación más detallada:
>>> df.pivot_table(rows=['A', 'B', 'C'])
A B C D_mean E_mean
0 bar one large 4.0 5.0
1 bar one small 5.0 3.0
...Use values para restringir las columnas que se van a agregar:
>>> df.pivot_table(rows=['A', 'B'], values='D')
A B D_mean
0 bar one 4.500000
1 bar two 6.500000
2 foo one 1.666667
3 foo two 3.000000Use aggfunc para aplicar una o más funciones de agregación:
>>> df.pivot_table(rows=['A', 'B'], values=['D'], aggfunc=['mean', 'count', 'sum'])
A B D_mean D_count D_sum
0 bar one 4.500000 2 9
1 bar two 6.500000 2 13
2 foo one 1.666667 3 5
3 foo two 3.000000 2 6Use columns para convertir los valores de una columna en nuevos encabezados de columna:
>>> df.pivot_table(rows=['A', 'B'], values='D', columns='C')
A B large_D_mean small_D_mean
0 bar one 4.0 5.0
1 bar two 7.0 6.0
2 foo one 2.0 1.0
3 foo two NaN 3.0Use fill_value para reemplazar NaN con un valor predeterminado:
>>> df.pivot_table(rows=['A', 'B'], values='D', columns='C', fill_value=0)
A B large_D_mean small_D_mean
0 bar one 4 5
1 bar two 7 6
2 foo one 2 1
3 foo two 0 3Conversión de cadenas clave-valor
DataFrame puede analizar cadenas clave-valor para separarlas en columnas distintas y convertir datos tabulares nuevamente en cadenas clave-valor. Para obtener información sobre la creación de objetos DataFrame, consulte Create a DataFrame object.
Extracción de pares clave-valor en columnas
Utilice extract_kv para analizar una columna que contiene cadenas clave-valor delimitadas:
>>> df
name kv
0 name1 k1=1,k2=3,k5=10
1 name1 k1=7.1,k7=8.2
2 name2 k2=1.2,k3=1.5
3 name2 k9=1.1,k2=1
>>> df.extract_kv(columns=['kv'], kv_delim='=', item_delim=',')
name kv_k1 kv_k2 kv_k3 kv_k5 kv_k7 kv_k9
0 name1 1.0 3.0 NaN 10.0 NaN NaN
1 name1 7.0 NaN NaN NaN 8.2 NaN
2 name2 NaN 1.2 1.5 NaN NaN NaN
3 name2 NaN 1.0 NaN NaN NaN 1.1Parámetros:
|
Parameter |
Description |
Default |
|
|
Campos de los cuales extraer pares clave-valor |
— |
|
|
Delimitador entre cada clave y su valor |
|
|
|
Delimitador entre pares clave-valor |
|
Los nombres de las columnas de salida siguen el patrón {original_field}_{key}, unidos con un guion bajo. Los valores faltantes toman NaN como predeterminado. Use fill_value para sustituir los valores faltantes por un valor predeterminado específico:
>>> df.extract_kv(columns=['kv'], kv_delim='=', fill_value=0)
name kv_k1 kv_k2 kv_k3 kv_k5 kv_k7 kv_k9
0 name1 1.0 3.0 0.0 10.0 0.0 0.0
1 name1 7.0 0.0 0.0 0.0 8.2 0.0
2 name2 0.0 1.2 1.5 0.0 0.0 0.0
3 name2 0.0 1.0 0.0 0.0 0.0 1.1Conversión de columnas a cadenas clave-valor
Utilice to_kv para serializar varias columnas en una sola columna de cadena clave-valor:
>>> df
name k1 k2 k3 k5 k7 k9
0 name1 1.0 3.0 NaN 10.0 NaN NaN
1 name1 7.0 NaN NaN NaN 8.2 NaN
2 name2 NaN 1.2 1.5 NaN NaN NaN
3 name2 NaN 1.0 NaN NaN NaN 1.1
>>> df.to_kv(columns=['k1', 'k2', 'k3', 'k5', 'k7', 'k9'], kv_delim='=')
name kv
0 name1 k1=1,k2=3,k5=10
1 name1 k1=7.1,k7=8.2
2 name2 k2=1.2,k3=1.5
3 name2 k9=1.1,k2=1