Python

Esta tarea te permite ejecutar scripts en lenguaje Python; la versión utilizada puede elegirse según las versiones puestas a disposición por tu administrador de Gaio.

Esta tarea te permite ejecutar scripts en lenguaje Python; la versión utilizada puede elegirse según las versiones puestas a disposición por tu administrador de Gaio. Las librerías pueden ser instaladas y gestionadas por los desarrolladores de Gaio. Además, proporcionamos una clase llamada bucket que te permite extraer y exportar datos que están en la base de datos ClickHouse a la que tu aplicación tiene permiso de acceder.

Límite de memoria La tarea de Python en Gaio está limitada por defecto a un máximo del 80% de la memoria de la máquina; si se supera este límite, se devolverá un error de límite de memoria.

---

Cómo configurar la tarea Python

Simplemente navegaremos por la interfaz de la tarea y, después, desarrollaremos un script simple como ejemplo.

---

1. Abrir la tarea Python

En el Studio, ve al panel Tareas. En la sección Analytics, selecciona Python.

---

2. Completar los campos requeridos

La primera pantalla es la principal de la tarea. En ella, a la izquierda, tenemos el espacio con tema azul para escribir el script, mientras que a la derecha, con tema oscuro, se encuentra la consola, donde podemos ver la salida del script. Para ejecutar tu script, simplemente haz clic en el botón Run y el resultado se mostrará en la consola.

Python Task Code Page

Es posible guardar los archivos generados en el script, como archivos jpeg, png, mp4, pkl, entre otros. El nombre de esta carpeta es assets.

Hay tres carpetas que puedes utilizar a través de la tarea de Python, que son las carpetas content, inputs y outputs de tu aplicación.

A continuación se muestra un ejemplo de cómo crear tu ruta hacia la carpeta outputs para poder descargar la imagen generada:

path = app_assets + "outputs/imagem_name.png"

En el cuadro de texto, debes escribir en cada línea el nombre correcto de la librería que deseas instalar (solo el nombre, sin ningún otro carácter, como se muestra en la imagen de abajo). Después de elegir la versión de Python y las librerías, simplemente haz clic en el botón Install para que se ejecuten tus configuraciones.

Python Task Environment Page

Como se mencionó anteriormente, tenemos una clase llamada bucket, que se conecta a ClickHouse de forma encapsulada y tiene los métodos query_df, command, insert_df y create_df.

Ejemplos

Función que transforma un SELECT de ClickHouse en un DataFrame de pandas en Python:

df = bucket.query_df('select columnA, columnB from table where columnB = 'active')

Función que hace una copia de una tabla de ClickHouse indicada a un DataFrame de pandas:

df = bucket . select_df ( 'new_table' )

En la primera línea tenemos la función que crea una tabla en ClickHouse similar a tu DataFrame de pandas; en la segunda línea insertamos los datos de tu DataFrame de pandas en la tabla de ClickHouse:

bucket . create_df ( 'new_table' , df )
bucket . insert_df ( 'new_table' , df )

Ten en cuenta que para ejecutar la función insert_df necesitamos que tu DataFrame de pandas sea similar a tu tabla de ClickHouse.

Ejemplo práctico

En este ejemplo práctico veremos la parte de traer los datos a Python, realizar un agrupamiento, guardar una imagen en formato png, guardar el archivo del modelo, y crear y guardar la tabla final en ClickHouse.

Primero, importemos las librerías que serán utilizadas:

import pandas as pd
from sklearn . cluster import KMeans
import matplotlib . pyplot as plt
import joblib

Para este ejemplo utilizaremos la famosa tabla iris proporcionada por varias librerías como scikit-learn. Esta tabla está en la base ClickHouse dentro de Gaio. Usaremos la función select_df para traerla a Python y luego aplicar el algoritmo K-Means proporcionado por la librería scikit-learn.

# Bring data into python
data = bucket . select_df ( 'iris_table' )

# Apply the K-Means algorithm with 3 clusters (number chosen arbitrarily)
kmeans = KMeans ( n_clusters =3 )
data [ 'cluster' ] = kmeans . fit_predict ( data )
​
# Evaluate the result - for example, viewing the means of each cluster
cluster_means = data . groupby ( 'cluster' ). mean ()

En el siguiente paso, visualizaremos los grupos encontrados por el modelo y guardaremos la figura en la carpeta assets.

# Plot the clusters on a graph (considering only the first two columns)
plt . scatter ( data [ 'sepal_length_cm_' ], data [ 'sepal_width_cm_' ], c = data [ 'cluster' ], cmap = 'viridis' )
plt . xlabel ( 'sepal_length_cm_' )
plt . ylabel ( 'sepal_width_cm_' )
​
# Save the chart in png format
plt . savefig ( 'assets/cluster_iris.png' )

Ahora guardaremos este modelo para reutilizarlo en otros momentos; para esto usaremos la librería joblib.

# Save the model
joblib . dump ( kmeans , 'assets/modelo_kmeans_iris.joblib' )

Ahora podemos enviar el DataFrame con la nueva columna generada por el modelo a ClickHouse para que pueda ser utilizado por otras tareas de Gaio. Para esto usaremos create_df e insert_df.

# Create a table in clickhouse similar to your dataframe
bucket . create_df ( 'tmp_iris_clusterizada' , data )
​
# Insert data from your dataframe into a clickhouse table
bucket . insert_df ( 'tmp_iris_clusterizada' , data )

Uso de Parámetros en Tareas Python

Las Python Tasks admiten parámetros dinámicos, lo que permite reutilizar el mismo script con diferentes entradas entre ejecuciones, entornos o flujos.

Los parámetros se resuelven en tiempo de ejecución y se inyectan automáticamente en el script.

---

1. Cómo funcionan los parámetros

Los parámetros definidos en la configuración de la tarea pueden referenciarse directamente dentro del código Python utilizando la siguiente sintaxis:

{{params.parameter_name}}

Durante la ejecución de la tarea, el valor del parámetro será sustituido automáticamente por el valor configurado.

---

2. Casos de uso comunes

Los parámetros pueden utilizarse en diferentes escenarios, como:

• Rutas dinámicas para la ingestión de archivos

• Selección de tablas o esquemas

• Lógica de ejecución condicional

• Configuraciones específicas por entorno

• Procesamientos basados en fechas

---

3. Notas importantes

• Los nombres de los parámetros son sensibles a mayúsculas y minúsculas (case-sensitive).

• Siempre valide los valores de los parámetros antes de utilizarlos en lógica crítica.

• Evite hardcoding de valores cuando los parámetros puedan utilizarse para hacer el script más flexible y reutilizable.

---

Uso de Tablas Temporales en Tareas Python

En las tareas Python de Gaio DataOS, es común crear tablas temporales para almacenar resultados intermedios durante el procesamiento de datos.

Para evitar conflictos de nombres entre ejecuciones simultáneas o entre diferentes usuarios, Gaio proporciona la función:

bucket.tmp_context_table_name(table_name)

Esta función genera automáticamente un nombre de tabla temporal con alcance de sesión, garantizando aislamiento entre ejecuciones.

---

Función

bucket.tmp_context_table_name(table_name)

Devuelve el nombre de una tabla temporal ajustado al contexto de ejecución.

Comportamiento

El nombre de la tabla puede variar dependiendo del tipo de contexto en el que se ejecute el script.

Contextos con sesión

Ejemplos: file-import, source, entre otros.

En estos casos, el nombre de la tabla recibe automáticamente un prefijo con el identificador de la sesión, garantizando aislamiento entre usuarios.

tmp_my_table → tmp_gaio{sessionId}_my_table

Contextos sin sesión

Ejemplos: studio, cron, rest, api.

En estos casos, el nombre de la tabla permanece sin cambios.

tmp_my_table → tmp_my_table

---

Cuándo utilizar

La función bucket.tmp_context_table_name() debe utilizarse siempre que sea necesario crear o referenciar tablas temporales (tmp_) dentro de scripts Python.

Esto evita:

• colisiones de nombres entre ejecuciones simultáneas

• interferencia entre usuarios

• conflictos entre pipelines concurrentes

---

Ejemplo de uso

import pandas as pd

df = pd.DataFrame({
    'col1': [1, 2],
    'col2': ['a', 'b']
})

table = bucket.tmp_context_table_name('tmp_my_table')

bucket.create_df(table, df)

result = bucket.query_df(f'SELECT * FROM {table}')

print(result)

En este ejemplo:

1. Se crea un DataFrame en memoria.

2. El nombre de la tabla temporal se genera con alcance de sesión.

3. El DataFrame se persiste en la tabla.

4. Los datos se consultan nuevamente desde la tabla creada.