Procesamiento de Streams de Datos
Tags: aws
Published: 2022-05-10 22:53:10

En este post vamos a ver como procesar un stream datos con Kinesis y Lambda.

Kinesis es un servicio que nos sirve para procesar datos en tiempo real. Algunas de sus caracteristicas son:

* Esta compuesto por Shard.Un Shard es la capacidad de datos que se pueden consumir, equivale a un 1 MB por segundo o mil registros por segundo.

* Los mensajes pueden ser almacenados entre 1 y 365 días.

* Dependiendo de la cantidad de data a consumir los shard se pueden de dividir (split) para tener más capacidad de consumo o se pueden unir (merge) para disminuir la capacidad de consumo.

* Partition Key es el identificador que se utiliza para decir en qué Shard se quiere almacenar el registro. Se define cuando se envía un mensaje a Kinesis.

Para este ejercicio vamos a utilizar el stream de wikipedia:[Stream](https://stream.wikimedia.org/v2/stream/recentchange)
qué va a ser procesado por Kinesis.

Empecemos creando un Data Stream. Vamos a buscar Kinesis en la barra de búsqueda, en la pantalla principal de Kinesis  seleccionamos _Kinesis Data Streams_ y damos click en _Create data stream_.

![Principal Kinesis]
(https://raw.githubusercontent.com/cecamilo00/posts/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/principal%20Kinesis.png)

Ahora configuremos nuestro data stream. Vamos a darle el nombre de: _Wikipedia-Stream_, seleccionamos el modo _Provisioned_  para aprovisionar un shard y damos click en el botón crear.  

![Create Kinesis 1]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/Create%20Kinesis%201.png?raw=true)

![Create Kinesis 2]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/Create%20Kinesis%202.png?raw=true)

Una vez creado nuestro data stream en la pestaña de configuración podemos modificar: la cantidad de shards aprovisionados, la encripción, el periodo de retención o si queremos tener métricas mejoradas en CloudWatch.

![Configuration Tab]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/Congifuration%20Tab.png?raw=true)

Ya que tenemos configurado nuestro data stream debemos enviarle la información. 
Vamos a utilizar el siguiente programa de Python para enviar los datos a Kinesis utilizando la librería boto3. El programa esta leyendo el siguiente stream de datos: _https://stream.wikimedia.org/v2/stream/recentchange_

```
import json
from sseclient import SSEClient as EventSource
import boto3
from datetime import datetime

clientKinesis = boto3.client('kinesis')
clientLambda = boto3.client('lambda')
url = 'https://stream.wikimedia.org/v2/stream/recentchange'

def get_object(change):
    return json.dumps({"id": change["id"],"bot":change["bot"],"length":change["length"]["new"],
                    "timestamp":str(datetime.fromtimestamp(change["timestamp"])),"type":change["type"],"user":change["user"],
                    "wiki":change["wiki"],"server_name":change["server_name"]})

def put_kinesis(data):
    return clientKinesis.put_record(
        StreamName='Wikipedia-Stream',
        Data=data,
        PartitionKey='1'
    )      


for event in EventSource(url):
    
    if event.event == 'message':
        try:
            change = json.loads(event.data)
            if ("id" in change and "bot" in change and "bot" in change and "length" in change 
                and "timestamp" in change and "type" in change
                and "user" in change and "wiki" in change and "server_name" in change): 
                put_kinesis(get_object(change))
        except ValueError:
            pass
```
Es necesario contar con las siguientes librerías instaladas para ejecutar el programa:

* boto3 pip install boto3
* sseclient pip install sseclient
* tener instalada y configurada la CLI

Ya que hemos configurado el productor de datos debemos configurar el consumidor que en este caso va a ser una función **Lambda**.

Vamos a buscar el servicio **Lambda** en el buscador y en la pagina principal damos click en el botón _Create function_. 

![Principal Lambda]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/Principal%20Lambda.png?raw=true)

Debemos definir el nombre de la función: lambda-stream, el runtime: python3.7 y dejamos el rol por defecto por ahora. 

![Principal Lambda]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/create%20function.png?raw=true)

Creada nuestra función Lamba debemos configurar el rol de la función para que pueda acceder a Kinesis. Para esto vamos a la pestaña configuración y damos click en el rol de la función.

![Edir Rol]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/edit%20rol.png?raw=true)

Debemos añadir la políticas necesarias: _AmazonKinesisFullAccess_ y la política de _CloudWatchFullAccess_ para poder monitorear los llamados a nuestra función.

![Policy Kinesis]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/policy%20kinesis.png?raw=true)

![Policy CloudWatch]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/policy%20cloudwatch.png?raw=true)

![Finish ROL]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/finish%20policies.png?raw=true)

Una vez creada nuestra función y asignado los permisos necesarios para su ejecución debemos configurar un trigger para que procese los datos de nuestro data stream.

![Add Trigger]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/add%20trigeer.png?raw=true)

EL siguiente paso es configurar el trigger, en donde el tipo de trigger es _Kinesis_, seleccionamos el nombre del data stream que ya hemos creado, definimos la cantidad de registros que queremos leer (en este caso vamos a leer de un solo registro) y definimos la posición en donde debemos comenzar a leer (Latest).

![Save Trigger]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/save%20trigger.png?raw=true)

Ya que terminamos la configuración vamos a modificar nuestra función Lambda para que imprima los datos que se reciben desde Kinesis.

```
import json
import base64


def lambda_handler(event, context):
    #Obtiene toda la infomación enviada por Kinesis
    print("EVENTO: " ,event)
    #Obtiene solamente los datos 
    print("DATA: ",json.loads(base64.b64decode(event["Records"][0]["kinesis"]["data"])))
    # TODO implement
    return {
        'statusCode': 200,
        'body': json.dumps('Kinesis!')
    }
```

El siguiente paso es ejecutar (_python3 nombrearchivo.py_) el programa para que empiece a enviar los datos a Kinesis.
Para ver los datos que procesa la función Lambda pestaña _Monitor_

![Monitor]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/configuration%20tab%20Lambda.png?raw=true)

En la siguiente pantalla vemos los datos que enviamos a imprimir en el log de **Cloud Watch**

![Log]
(https://github.com/cecamilo00/posts/blob/circleci-project-setup/Procesamiento%20de%20Streams%20de%20Datos/log%20lambda.png?raw=true)

Ahora que sabemos como obtener los datos que enviamos a Kinesis los invito a guardar esta información en DynamoDB, RDS o procesarla con otro de servicio.


Me pueden encontrar en:

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

# Referencias

[Kinesis](https://aws.amazon.com/es/kinesis/?nc=sn&loc=0)

[Install Boto3](https://pypi.org/project/boto3/)

[Install sseclient](https://pypi.org/project/sseclient/)

[Install CLI](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html)

[DynamoDb TTL](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/TTL.html)