Paso 2. Implementar la lógica de transformación - AWS Glue

Paso 2. Implementar la lógica de transformación

nota

Las transformaciones visuales personalizadas solo admiten scripts de Python. No se admite Scala.

Para agregar el código que implementa la función definida mediante el archivo de configuración .json, se recomienda colocar el archivo de Python en la misma ubicación que el archivo .json, con el mismo nombre, pero con la extensión “.py”. AWS Glue Studio empareja automáticamente los archivos .json y .py para que no tenga que especificar la ruta del archivo Python en el archivo de configuración.

En el archivo de Python, agregue la función declarada, con los parámetros con nombre configurados y regístrela para que se pueda utilizar en DynamicFrame. A continuación, se muestra un ejemplo de un archivo de Python: 

from awsglue import DynamicFrame

# self refers to the DynamicFrame to transform, 
# the parameter names must match the ones defined in the config
# if it's optional, need to provide a default value
def myTransform(self, email, phone, age=None, gender="", 
                      country="", promotion=False):
   resulting_dynf = # do some transformation on self 
   return resulting_dynf
   
DynamicFrame.myTransform = myTransform

Se recomienda utilizar un cuaderno de AWS Glue para desarrollar y probar el código de Python de la forma más rápida. Consulte Introducción a los cuadernos en AWS Glue Studio

Para ilustrar cómo implementar la lógica de transformación, la transformación visual personalizada del ejemplo siguiente es una transformación para filtrar los datos entrantes y conservar solo los datos relacionados con un estado concreto de EE. UU. El archivo .json contiene el parámetro para functionName como custom_filter_state y dos argumentos (“state” y “colName” con el tipo “str”).

El archivo .json de configuración de ejemplo es el siguiente: 

{
"name": "custom_filter_state",
"displayName": "Filter State",
"description": "A simple example to filter the data to keep only the state indicated.",
"functionName": "custom_filter_state",
"parameters": [
   {
    "name": "colName",
    "displayName": "Column name",
    "type": "str",
    "description": "Name of the column in the data that holds the state postal code"
   },
   {
    "name": "state",
    "displayName": "State postal code",
    "type": "str",
    "description": "The postal code of the state whole rows to keep"
   }   
  ]
}
Implementación del script complementario en Python

  1. Inicie un cuaderno de AWS Glue y ejecute la celda inicial proporcionada para iniciar la sesión. Al ejecutar la celda inicial, se crean los componentes básicos necesarios.

  2. Cree una función que filtre tal y como se describe en el ejemplo y regístrela en DynamicFrame. Copie el código siguiente y péguelo en una celda del cuaderno de AWS Glue. 

    from awsglue import DynamicFrame

def custom_filter_state(self, colName, state):
    return self.filter(lambda row: row[colName] == state)

DynamicFrame.custom_filter_state = custom_filter_state

  3. Cree o cargue datos de muestra para probar el código en la misma celda o en una nueva. Si agrega los datos de muestra en una celda nueva, no olvide ejecutar la celda. Por ejemplo: 

    # A few of rows of sample data to test
data_sample = [
    {"state": "CA", "count": 4},
    {"state": "NY", "count": 2},
    {"state": "WA", "count": 3}    
]
df1 = glueContext.sparkSession.sparkContext.parallelize(data_sample).toDF()
dynf1 = DynamicFrame.fromDF(df1, glueContext, None)

  4. Haga pruebas para validar “custom_filter_state” con distintos argumentos:

    En la captura de pantalla, se muestra una celda de un cuaderno de AWS Glue con los argumentos pasados a la función DynamicFrame.show.

  5. Tras ejecutar varias pruebas, guarde el código con la extensión .py y asigne un nombre al archivo .py que refleje el nombre del archivo .json. Los archivos .py y .json deben estar en la carpeta de la misma transformación.

    Copie el siguiente código, péguelo en un archivo y cámbiele el nombre con una extensión de archivo .py. 

    from awsglue import DynamicFrame

def custom_filter_state(self, colName, state):
    return self.filter(lambda row: row[colName] == state)

DynamicFrame.custom_filter_state = custom_filter_state

  6. En AWS Glue Studio, abra un trabajo visual y agregue la transformación al trabajo; para ello, selecciónela en la lista Transforms (Transformaciones) disponibles.

    Para reutilizar esta transformación en el código de un script de Python, agregue la ruta de Amazon S3 al archivo .py en el trabajo, en “Referenced files path” (Ruta de archivos a la que se hace referencia), y en el script. Por último, importe el nombre del archivo de Python (sin la extensión); para ello, agréguelo a la parte superior del archivo. Por ejemplo: import <nombre del archivo (sin la extensión)>