MaxFrame es una API de DataFrame compatible con Pandas para MaxCompute. Le permite ejecutar operaciones de datos al estilo de Pandas sobre conjuntos de datos a gran escala, sin las limitaciones de memoria o recursos de cómputo locales. Utilice MaxFrame cuando su conjunto de datos supere la memoria de una sola máquina o cuando sus flujos de trabajo de Pandas sean demasiado lentos para ejecutarse de manera local.
Para migrar desde Pandas, solo cambie una importación. MaxFrame se encarga del resto y distribuye el procesamiento en el backend de MaxCompute:
# Before (pandas)
import pandas as pd
# After (MaxFrame)
import maxframe.dataframe as mdMaxFrame utiliza evaluación diferida. Las operaciones construyen un grafo de cómputo, pero no se ejecutan hasta que usted llama a.execute(). Encadene todas las transformaciones que necesite y después llame a.execute()una sola vez para iniciar el procesamiento.
Requisitos previos
Antes de empezar, asegúrese de contar con lo siguiente:
Un proyecto de MaxCompute con los permisos obligatorios
Su ID de clave de acceso y su clave de acceso secreta de Alibaba Cloud (configurados como variables de entorno)
MaxFrame instalado en su entorno de Python
Conjunto de datos de ejemplo
Esta guía utiliza la tabla maxframe_ml_100k_users del conjunto de datos público de MaxCompute. La tabla está disponible en BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users, por lo que no necesita importarla.
Inicializar una sesión
Antes de ejecutar cualquier tarea de MaxFrame, inicialice una sesión en el punto de entrada de su código. Todo el procesamiento de datos posterior pasa por esta sesión, la cual se comunica con el backend de MaxCompute.
import os
from maxframe import new_session
from odps import ODPS
# Initialize MaxCompute with your credentials.
o = ODPS(
os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
project='your-default-project',
endpoint='your-end-point',
)
# Start the MaxFrame session.
new_session(odps_entry=o)Para obtener más información sobre los parámetros, consulte MaxFrame-specific APIs.
Crear un DataFrame
Utilice read_odps_table o read_odps_query para cargar datos de MaxCompute en un DataFrame de MaxFrame. También puede crear un DataFrame a partir de datos locales para pruebas rápidas.
Leer desde una tabla de MaxCompute
La función read_odps_table lee una tabla de MaxCompute y devuelve un DataFrame.
import maxframe.dataframe as md
# Read all columns from the table.
df = md.read_odps_table('BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users')Use el parámetro index_col para establecer una columna como índice del DataFrame, o bien columns para seleccionar un subconjunto de columnas:
# Set user_id as the index.
df = md.read_odps_table(
'BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users',
index_col='user_id'
)
# Select specific columns.
df = md.read_odps_table(
'BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users',
columns=['user_id', 'age', 'sex']
)Para conocer todos los detalles de los parámetros, consulte MaxFrame-specific APIs.
Leer desde una consulta SQL
Ejecute una consulta SQL y utilice el resultado como un DataFrame:
import maxframe.dataframe as md
df = md.read_odps_query(
'select user_id, age, sex FROM `BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users`'
)
# Optionally set the index column.
df = md.read_odps_query(
'select user_id, age, sex FROM `BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users`',
index_col='user_id'
)Para conocer todos los detalles de los parámetros, consulte MaxFrame-specific APIs.
Crear desde datos locales
Para hacer pruebas locales rápidas, cree un DataFrame directamente desde un diccionario de Python:
import maxframe.dataframe as md
d = {'col1': [1, 2], 'col2': [3, 4]}
df = md.DataFrame(data=d)
print(df.execute().fetch())
# Result:
# col1 col2
# 0 1 3
# 1 2 4Procesar datos
MaxFrame admite las mismas operaciones de datos que Pandas. Todas utilizan evaluación diferida: encadene las que necesite y luego llame a .execute() una sola vez para iniciar el procesamiento.
Operaciones matemáticas
Aplique operaciones aritméticas por columna. Todos los operadores estándar (+, -, *, /) funcionan igual que en Pandas.
import maxframe.dataframe as md
df = md.DataFrame({'angles': [0, 3, 4],
'degrees': [360, 180, 360]},
index=['circle', 'triangle', 'rectangle'])
# Addition
print((df + 1).execute().fetch())
# Result:
# angles degrees
# circle 1 361
# triangle 4 181
# rectangle 5 361
# Multiplication
print((df * 2).execute().fetch())
# Result:
# angles degrees
# circle 0 720
# triangle 6 360
# rectangle 8 720
# Division
print((df / 2).execute().fetch())
# Result:
# angles degrees
# circle 0.0 180
# triangle 1.5 90
# rectangle 2.0 180
# Exponentiation
print((df ** 2).execute().fetch())
# Result:
# angles degrees
# circle 0 129600
# triangle 9 32400
# rectangle 16 129600Para obtener más información, consulte Binary operator functions.
Filtrado, proyección y muestreo
Filtrado
Seleccione las filas que cumplan con una condición:
import maxframe.dataframe as md
df = md.read_odps_table('BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users')
# Keep only rows where age > 18.
filtered_df = df[df['age'] > 18]
print(filtered_df.execute().fetch())
# Result:
# user_id age sex occupation zip_code
# 0 1 24 M technician 85711
# 1 2 53 F other 94043
# ...Proyección
Seleccione un subconjunto de columnas:
import maxframe.dataframe as md
df = md.read_odps_table('BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users')
# Select specific columns.
projected_df = df[['user_id', 'age']]
print(projected_df.execute().fetch())
# Result:
# user_id age
# 0 1 24
# 1 2 53
# ...Muestreo
Extraiga una muestra aleatoria de un DataFrame:
import maxframe.dataframe as md
df = md.DataFrame({'num_legs': [2, 4, 8, 0],
'num_wings': [2, 0, 0, 0],
'num_specimen_seen': [10, 2, 1, 8]},
index=['falcon', 'dog', 'spider', 'fish'])
print(df['num_legs'].sample(n=3, random_state=1).execute())
# Result:
# falcon 2
# fish 0
# dog 4
# Name: num_legs, dtype: int64Para obtener más información, consulte Reindexing / selection / label manipulation.
Ordenamiento
Ordene un DataFrame por una o más columnas mediante sort_values:
import maxframe.dataframe as md
import numpy as np
df = md.DataFrame({
'col1': ['A', 'A', 'B', np.nan, 'D', 'C'],
'col2': [2, 1, 9, 8, 7, 4],
'col3': [0, 1, 9, 4, 2, 3],
})
# Sort by col1.
res = df.sort_values(by=['col1']).execute()
print(res.fetch())
# Result:
# col1 col2 col3
# 0 A 2 0
# 1 A 1 1
# 2 B 9 9
# 5 C 4 3
# 4 D 7 2
# 3 None 8 4
# Sort by col1, then col2.
res = df.sort_values(by=['col1', 'col2']).execute()
print(res.fetch())
# Result:
# col1 col2 col3
# 1 A 1 1
# 0 A 2 0
# 2 B 9 9
# 5 C 4 3
# 4 D 7 2
# 3 None 8 4Para obtener más información, consulte Reshaping / sorting / transposing.
Unión, combinación y concatenación
Combine DataFrames de forma horizontal (join/merge) o vertical (concatenación).
import maxframe.dataframe as md
df1 = md.DataFrame({'key': ['K0', 'K1', 'K2'],
'A': ['A0', 'A1', 'A2'],
'B': ['B0', 'B1', 'B2']})
df2 = md.DataFrame({'key': ['K0', 'K1', 'K2'],
'C': ['C0', 'C1', 'C2'],
'D': ['D0', 'D1', 'D2']})
# Merge on a key column.
result = md.merge(df1, df2, on='key')
print(result.execute().fetch())
# Result:
# key A B C D
# 0 K0 A0 B0 C0 D0
# 1 K1 A1 B1 C1 D1
# 2 K2 A2 B2 C2 D2
# Concatenate vertically.
result = md.concat([df1, df2])
print(result.execute().fetch())Para obtener más información, consulte Combining / joining / merging.
Agregación y funciones definidas por el usuario (UDF)
Agregación
Resuma los datos por grupos o columnas con funciones de agregación:
import maxframe.dataframe as md
import numpy as np
df = md.DataFrame([[1, 2, 3],
[4, 5, 6],
[7, 8, 9],
[np.nan, np.nan, np.nan]],
columns=['A', 'B', 'C'])
# Apply multiple aggregation functions.
print(df.agg(['sum', 'min']).execute().fetch())
# Result:
# A B C
# min 1.0 2.0 3.0
# sum 12.0 15.0 18.0
# Apply different functions to different columns.
df.agg({'A': ['sum', 'min'], 'B': ['min', 'max']}).execute().fetch()
# Result:
# A B
# max NaN 8.0
# min 1.0 2.0
# sum 12.0 NaNUDF
MaxFrame admite UDF mediante transform y apply. Utilícelas cuando necesite lógica personalizada que los operadores integrados no cubren.
Antes de ejecutar una UDF, declare la imagen común. Para ello, configure config.options.sql.settings antes de llamar a new_session.
Todos los ejemplos de UDF utilizan este bloque de configuración:
from maxframe import config
config.options.sql.settings = {
"odps.session.image": "common",
"odps.sql.type.system.odps2": "true"
}
session = new_session(o)Utilice transform para aplicar una UDF por columna (la forma de salida coincide con la de entrada):
import maxframe.dataframe as md
df = md.DataFrame({'A': range(3), 'B': range(1, 4)})
print(df.transform(lambda x: x + 1).execute().fetch())
# Result:
# A B
# 0 1 2
# 1 2 3
# 2 3 4Utilice apply cuando la cantidad de columnas de salida difiera de la de entrada:
import maxframe.dataframe as md
import numpy as np
def simple(row):
row['is_man'] = row['sex'] == "man"
return row
df = md.read_odps_table('BIGDATA_PUBLIC_DATASET.data_science.maxframe_ml_100k_users')
new_dtypes = df.dtypes.copy()
new_dtypes["is_man"] = np.dtype(np.bool_)
df.apply(
simple,
axis=1,
result_type="expand",
output_type="dataframe",
dtypes=new_dtypes
).execute().fetch()
# Result:
# user_id age sex occupation zip_code is_man
# 0 1 24 M technician 85711 False
# 1 2 53 F other 94043 False
# ...
# [943 rows x 6 columns]Para obtener más información, consulte Function application / GroupBy / window.
Almacenar resultados
Una vez procesados los datos, guárdelos en una tabla de MaxCompute con to_odps_table.
# Write results to a MaxCompute table.
filtered_df.to_odps_table('<table_name>')
# Set a lifecycle (in days) for the result table.
filtered_df.to_odps_table('<table_name>', lifecycle=7)|
Parameter |
Description |
|
|
Nombre de la tabla de destino. Si la tabla no existe, MaxCompute la crea automáticamente. Si ya existe, los datos se anexan de forma predeterminada. |
|
|
Número de días que se conservará la tabla de resultados. |
|
|
Establezca este parámetro en |
Para obtener más información, consulte MaxFrame-specific APIs.
Ejecutar y recuperar resultados
MaxFrame utiliza evaluación diferida: las operaciones construyen un grafo de cómputo, pero no se ejecutan hasta que usted llama a .execute(). Encadene .execute().fetch() para iniciar la ejecución y recuperar los resultados.
# Execute the computation and fetch results.
data = filtered_df.execute().fetch()
print(data)|
Method |
Description |
|
|
Envía el grafo de cómputo a MaxCompute y espera a que finalice. |
|
|
Recupera una parte de los datos resultantes después de la ejecución. |
|
|
Inicia la ejecución y devuelve los resultados en un solo paso. |
Para obtener más información, consulte MaxFrame-specific APIs y MaxFrame-specific APIs.