Limpiar y preparar grandes cantidades de datos de manera fácil con Glue DataBrew
Tags: aws
Published: 2023-03-27 16:48:49

Usualmente cuando queremos realizar un proceso de ETL (Extraer, Transformar, Cargar) utilizamos algunas herramientas o lenguajes de programación nos ayudan en este proceso. El desarrollo de ETL requiere de ciertos conocimientos técnicos, que nos tomaran un tiempo en adquirir. Para simplificar este proceso AWS tiene un servicio llamado [Glue DataBrew](https://aws.amazon.com/es/glue/features/databrew/) que nos va a simplificar y agilizar la tarea de realizar las ETL.

En este post vamos a crear la siguiente arquitectura:

![Arquitectura](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/arquitecturagluedatabrew.png?raw=true)

EL objetivo de esta arquitectura es: Cuando se cargue un archivo al Bucket este llame a una función Lambda que a su vez va a ejecutar una Step Fucntion que va realizar el procesamiento que hallamos definido para nuestro data set y enviara un mensaje de texto si se ejecuto correctamente o si hubo un error durante el procesamiento de la información.  

Iniciemos creando el Bucket que va almacenar nuestro conjunto de datos y el resultado del proceso de ETL. Para esto buscamos el servicio de S3 y damos click en el botón _Create Bucket_, una vez en la pagina de creación del Bucket le damos un nombre (el nombre debe ser úrico en todo AWS), dejamos las opciones por defecto y damos click en el botón en _Create Bucket_ que se encuentra en la parte inferior de la pantalla.

![Crear Bucket](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/bucketgluedatabrew1.png?raw=true)

Vamos a crear dos carpetas en nuestro Bucket una para los archivos que vamos a procesar y otra de resultados. Damos click en el botón _Create Folder_.

![Create Folder](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/createfolders3.png?raw=true)

La carpeta que contiene el archivo a procesar le damos el nombre de _procesardata_ y la que almacena los _resultados_ le damos el nombre de resultados. 
Creamos la carpeta _procesardata_ y realizamos el mismo procedimiento para la carpeta de _resultados_.

![Create Procesardata](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/folderprocesardata.png?raw=true)

Creadas las carpetas deberíamos verlas en nuestro Bucket.

![Folders](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/carpetas%20s3.png?raw=true)

Ahora necesitamos subir a la carpeta de _procesardata_ el archivo con el cual vamos a trabajar damos click en la carpeta y damos click en el botón _Upload_. 

![Folders](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/upload%20file.png?raw=true)

Seleccionamos el [archivo](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/precios.csv) y damos click en el botón _Upload_.

Ya que tenemos el archivo con el cual vamos a trabajar, buscamos el servicio **Glue DataBrew** una vez en la pagina principal damos click en _DataSets_.

![Page DataSet](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/page%20data%20set.png?raw=true)

En la siguiente pantalla seleccionamos el Bucket, el folder _procesardata_ y el archivo _precios.csv_.

![DataSet](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/select%20dataset%20file.png?raw=true)

Para terminar con la creación del DataSet seleccionamos el tipo de archivo el cual estamos utilizando (.CSV separado por comas) e indicamos que la primera fila del archivo es la cabecera y damos click en el botón _Create dataset_.

![Type File](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/file%20type.png?raw=true)

Creado nuestro DataSet el siguiente paso es crear un proyecto  en el cual vamos a realizar las modificaciones de nuestros datos. para esto damos click en _Projects_ y luego en el botón _Create project_.

![Page Project](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/page%20project.png?raw=true)

Para crear el proyecto necesitamos la siguiente información:

- Nombre del proyecto

- Receta o nombre de la receta si vamos a crear una nueva. La receta son las modificaciones que vamos a realizar al conjunto de datos.

- Debemos seleccionar el Dataset con el cual vamos a trabajar.

- Debemos utilizar un rol existente o crear uno nuevo definiendo un prefijo para este.

![Create Project](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/crear%20proyecto.png?raw=true)

Creado nuestro proyectos veremos una pagina como la siguiente. 

1. En la parte superior vamos a ver las opciones de uso frecuente de la receta que podemos aplicar a nuestros datos. 
2. Abajo de la cabecera de nuestros datos vamos a poder observar las estadísticas de cada columna, lo que nos ayuda a tener una idea del comportamiento de los datos que están almacenados dentro de cada una de ellas.
3. En la parte derecha de la pantalla tenemos la sección de la receta, donde vamos a poder agregar diferentes tipos de modificaciones a nuestro conjunto de datos. 

![Page Project](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/pagina%20proyecto%201.png?raw=true)

Ya que esta listo nuestro proyecto para trabajar vamos a crear nuestra receta, para esto damos click en el botón _Add step_ que se encuentra en la sección de la receta.

![Add step](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/add%20step.png?raw=true)

Una vez damos click en el botón _Add step_ se despliegan las diferentes opciones que podemos utilizar para modificar los datos. Vamos a seleccionar la opción _Change to capital case_ de la sección de FORMAT.

![Capital Case 1](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/capital%20case%201.png?raw=true)

Como la columna PRODUCTO tiene los nombres en mayusculas los vamos a cambiar para que cada palabra empieza por una letra en mayúscula.

![Capital Case 2](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/capital%20case%202.png?raw=true)

Al dar click en el botón Apply veremos los resultados.

![Capital Case 3](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/result%20capital%20case.png?raw=true)

Vemos que las columnas de los días tienen los precios de cada producto, sin embargo si quisiéramos hacer alguna operación matemática sobre ellos no será posible debido a que tienen el signo $. Veamos como podemos eliminar el signo $ de las columnas. 

Vamos al menú superior seleccionamos _Clean_ y en la sección _Remove_ seleccionamos _Special characters_.

![Remove Special Characters](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/remove%20special%20characters%201.png?raw=true)

Al realizar la selección en sección de _Recipies_ se nos despliegan las opciones para realizar la limpieza de la columna. Seleccionamos la columna _LUNES_,_Special characters_ y _Custom value_ para remover el signo - que aparece delante de los ceros. 


![Remove Special Characters 2](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/remove%20special%20characters%202.png?raw=true)

Al dar click en el botón _Apply_ veremos que de la columna LUNES se quitaron los caracteres: - $ y se agrego un nuevo paso a nuestra receta. Este proceso lo puedes realizar para el resto de columnas de los días de la semana.

![Remove Special Characters 3](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/remove%20special%20characters%20%203.png?raw=true)

Al finalizar de remover los caracteres especiales de las columnas de los días de la semana, nuestra receta se vera de la siguiente forma.

![Remove Special Characters 4](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/remove%20special%20%20characters%204.png?raw=true)

Ahora que nuestras columnas no tienen caracteres especiales vamos a cambiar el tipo de dato para poder realizar algunas operaciones. En la parte superior de cada columna podemos ver el tipo de datos de cada columna, en las columnas de los días de la semana identificamos que son de tipo string y necesitamos cambiarlas a un valor numérico en este caso double.

![Change Type 1](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/change%20type%201.png?raw=true)

Al dar click en _Change type_ se nos presentan las opciones donde debemos seleccionar la columna y el tipo de dato al cual queremos cambiar.

![Change Type 2](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/change%20type%202.png?raw=true)

Damos click en el botón _Apply_ dando como resultado el cambio de tipo de columna y un nuevo paso se agrego a nuestra receta. 

![Change Type 3](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/change%20type%203.png?raw=true)

Debemos hacer este mismo procedimiento a las otras columnas, para que nuestra receta quede de la siguiente forma.

![Change Type 4](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/change%20type%204.png?raw=true)

Para terminar con nuestra receta vamos a crear un nueva columna llamada Total que va almacenar la suma de los días _LUNES_ Y _VIERNES_.
Hacemos click en botón _FUNCTIONS_ - Math functions - _ADD_. 

![Sum1](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/sum%201.png?raw=true)

En la sección de la receta veremos las opciones donde seleccionamos el tipo de función (ADD), las columnas que vamos a sumar (LUNES Y VIERNES) y por ultimo damos el nombre de la columna que va almacenar el resultado (Total).

![Sum2](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/sum%202.png?raw=true)

Al dar click en el botón _Apply_ vemos que se creo la columna total y contiene la suma de las columnas _LUNES_ Y _VIERNES_.

![Sum3](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/sum%203.png?raw=true)

Con esto finalizamos la modificación de nuestro DataSet, ahora necesitamos definir como se van a ejecutar estas modificaciones por lo tanto necesitamos crear un Job.
Damos click en el botón _Create job_.

![Select Job](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/select%20job.png?raw=true)

En la pantalla de creación del Job le asignamos un nombre al Job, seleccionamos el Bucket y la carpeta en la cual queremos que se almacenen los resultados (Bucket y carpeta que creamos al inicio) y le asignamos un rol que puede ser el que creamos  al momento de configurar el proyecto o uno nuevo.

![Create Job](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/create%20job.png?raw=true)

Al crearse el Job vamos a ver el listado con los Jobs que hayamos creado, seleccionamos el Job que acabamos de crear y damos click en el botón _Run Job_.

![Run Job](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/list%20job%202.png?raw=true)

Para ver los resultados podemos ir a la carpeta _resultados_ del Bucket que creamos y ver el archivo que genero el Job.

![Results Job](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/job%20results.png?raw=true)

Hemos visto como crear, modificar un DataSet y obtener los resultados por medio de las recetas y los Jobs. 
Ahora necesitamos implementar los siguientes servicios para completar nuestra arquitectura:
  
- Un Topic y una Suscripción de SNS para enviar los mensajes de notificación.

- Una Step Function que va a realizar todo el flujo para procesar el Dataset, modificando el Dataset para que tome el archivo que se suba a S3, ejecutar el Job y enviar una mensaje de texto para notificar que el proceso se realizo correctamente.

- Una función Lambda que se va encargar de llamar a la Step Function cuando se suba un archivo al Bucket de S3.

Esta implementación la vamos a realizar con la siguiente [plantilla](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/gluedatabrew.tf) de Terraform.

```
provider "aws" {
  region = "us-east-1"
}


resource "random_id" "random" {
  byte_length = 2
}

variable "job_name"{
    type= string
    default= "PreciosJob"
}

variable "bucket_name" {
    type = string
    default = "datosgluedatabrew"
}

variable "telefono"{
    type = string
    default = "+573005465765"
}

resource "aws_lambda_function" "lambda_function_call_step_function" {
  filename      = "lambda_function.py.zip"
  function_name = "LambdaStepFunction-${random_id.random.hex}"
  role          = aws_iam_role.lambda_execution_role.arn
  handler       = "lambda_function.lambda_handler"
  runtime       = "python3.8"
  memory_size   = 128
  timeout       = 5
  environment {
    variables = {
      STEPFUNCTIONARN = aws_sfn_state_machine.function.arn
    }
  }
}

resource "aws_iam_role" "lambda_execution_role" {
  name = "lambda_execution_role-${random_id.random.hex}"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "lambda.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy_attachment" "lambda_execution_role_policy_attach" {
  policy_arn = "arn:aws:iam::aws:policy/AWSStepFunctionsFullAccess"
  role       = aws_iam_role.lambda_execution_role.name
}

resource "aws_iam_role_policy_attachment" "lambda_execution_role_cloudwatch_attach" {
  policy_arn = "arn:aws:iam::aws:policy/CloudWatchLogsFullAccess"
  role       = aws_iam_role.lambda_execution_role.name
}


resource "aws_lambda_permission" "s3_permission_to_trigger_lambda" {
  statement_id  = "AllowExecutionFromS3Bucket"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.lambda_function_call_step_function.arn
  principal     = "s3.amazonaws.com"
  source_arn    = "arn:aws:s3:::${var.bucket_name}"
}

resource "aws_s3_bucket_notification" "bucket_notification_put" {
  bucket = "${var.bucket_name}"

  lambda_function {
    lambda_function_arn = aws_lambda_function.lambda_function_call_step_function.arn
    events              = ["s3:ObjectCreated:Put"]
    filter_prefix = "procesardata/"
    filter_suffix = ".csv"
  }
  depends_on = [aws_lambda_permission.s3_permission_to_trigger_lambda]
}


resource "aws_sfn_state_machine" "function" {
  name     = "StateMachineGlueDataBrew-${random_id.random.hex}"
  role_arn = aws_iam_role.function_role.arn

  definition = <<DEFINITION
    {
  "Comment": "A description of my state machine",
  "StartAt": "Actualizar DataSet",
  "States": {
    "Actualizar DataSet": {
      "Type": "Task",
      "Parameters": {
        "Format": "CSV",
        "Input": {
          "S3InputDefinition": {
            "Bucket.$": "$.bucket",
            "BucketOwner.$": "$.cuenta",
            "Key.$": "$.carpeta"
          }
        },
        "Name": "Precios"
      },
      "Resource": "arn:aws:states:::aws-sdk:databrew:updateDataset",
      "Next": "Ejecutar Job"
    },
    "Ejecutar Job": {
      "Type": "Task",
      "Resource": "arn:aws:states:::databrew:startJobRun",
      "Parameters": {
        "Name": "${var.job_name}"
      },
      "Catch": [
        {
          "ErrorEquals": [
            "DataBrew.ConflictException"
          ],
          "Next": "SNS Error"
        }
      ],
      "Next": "SNS Job Ejecutado"
    },
    "SNS Error": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "Message": {
          "Mensaje": "Se produjo un error"
        },
        "TopicArn": "${aws_sns_topic.topic_glue_databrew.arn}"
      },
      "End": true
    },
    "SNS Job Ejecutado": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "${aws_sns_topic.topic_glue_databrew.arn}",
        "Message": {
          "Mensaje": "Se ejecuto el job correctamente"
        }
      },
      "End": true
    }
  }
}
  DEFINITION
}

resource "aws_iam_role" "function_role" {
  name = "my-state-machine-role-${random_id.random.hex}"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "states.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy_attachment" "function_policy_role_sns" {
  policy_arn = "arn:aws:iam::aws:policy/AmazonSNSFullAccess"
  role       = aws_iam_role.function_role.name
}

resource "aws_iam_role_policy_attachment" "lambda_policy_glue_data_brew" {
  policy_arn = "arn:aws:iam::aws:policy/AwsGlueDataBrewFullAccessPolicy"
  role       = aws_iam_role.function_role.name
}

resource "aws_sns_topic" "topic_glue_databrew" {
  name = "topic_glue_databrew-${random_id.random.hex}"
}

resource "aws_sns_topic_subscription" "subscription_glue_databrew" {
  topic_arn = aws_sns_topic.topic_glue_databrew.arn
  protocol  = "sms"
  endpoint  = "${var.telefono}"
}

```
De la plantilla anterior debemos modificar las variables y la región en la cual se van desplegar los servicios. 

La región.
```
provider "aws" {
   region = "us-east-1"
 }
```
El nombre que le dimos la Job.
```
variable "job_name"{
    type= string
    default= "PreciosJob"
}
```
El nombre que le dimos al Bucket.
```
variable "bucket_name" {
    type = string
    default = "datosgluedatabrew"
}
```
El numero de teléfono al cual queremos que lleguen los mensajes cuando se ejecuta el Job.
```
variable "telefono"{
    type = string
    default = "+573005465765"
}
```
Una vez que hemos modificado las variables de acuerdo a nuestras configuraciones podemos validar y ejecutar la plantilla.

- Valida la plantilla
```
terraform validate
```
- Crea los servicios. Recuerda que la [plantilla](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/gluedatabrew.tf) y el [código](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/lambda_function.py.zip) de la función _Lambda_ se encuentren en la misma carpeta.

```
terraform apply
```

Una vez terminada la ejecución del comando, vamos a revisar que los servicios se hayan desplegado correctamente.

Buscamos _SNS_ y damos click en _Topics_.

![SNS](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/create%20topic.png?raw=true)

Buscamos la función _Lambda_.

![Lambda](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/create%20lambda.png?raw=true)

Buscamos la _Step Function_.

![Step Function](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/create%20step%20function.png?raw=true)

Como todos los servicios fueron desplegados con éxito vamos a subir un nuevo archivo al _Bucket_ para que se procese el archivo con la llamada a la  función _Lambda_ y la _Step Function_, cuando esta termina su proceso damos click en el nombre de la _Step Function_  y damos click en la ultima ejecución para ver el gráfico del proceso.

![ Exec Step Function](https://github.com/cecamilo00/posts/blob/circleci-project-setup/gluedatabrew/exec%20step%20funcion.png?raw=true)

Con esto finalizamos la implementación y pruebas de la arquitectura planteada al inicio.Pueden realizar mas pruebas modificando la receta o agregando mas pasos a la _Step Funcion_.

Para no incurrir en ningún costo pueden borrar el proyecto de _Glue DataBrew_, el Bucket y para eliminar los servicios creados con _Terraform_ pueden utilizar el siguiente comando.

```
terraform destroy
```

Me pueden encontrar en

[Camilo Cabrales](https://www.linkedin.com/in/camilo-cabrales-31a0bb53/)

# Referencias

[API](https://docs.aws.amazon.com/databrew/latest/dg/api-reference.html)