En este primer capítulo, conocerás las diferencias entre el procesamiento por lotes y por flujos, crearás tu primer flujo, gestionarás sus permisos, escribirás en él y leerás de él.

¿Qué es el streaming y por qué es importante?

Lote vs flujo

Productores y destinos

Gestionar flujos de entrega Firehose

Preparándose para el primer flujo

Crear roles

IAM usuarios vs. roles

Práctica de permisos

Creación de bucket S3

Trabajar con el flujo de entrega Firehose

Crea tu primer flujo Firehose

Escribir en un flujo Firehose

Lectura de datos Firehose

Streaming en la nube

El siguiente paso en tu viaje por el streaming de datos es aprender a utilizar funciones lambda transformacionales para ir sin servidor. Mediante ejercicios prácticos, añadirás capas lambda y activarás funciones lambda en condiciones específicas.

Sin servidores

Crear y ejecutar funciones Lambda

Opciones

Genial para un servidor

Genial para Serverless

Sin servidor vs servidores

Disparador

Manipulador

Capa

Componentes de la función lambda

# EDIT ME!
# recordReaderS3/lambda_function.py
# This is lambda handler code that will execute each time an object is added to S3.
import json

def record_created_handler(____, ____):
    # Create a list to store new object keys
    written_objects = []

    # Iterate over each S3 event record
    for record in event['Records']:
       
        # Get the variables to check for
        event_name = record['eventName']
        bucket_name = record['s3']['bucket']['name']
        obj_key = record['s3']['object']['key']

        # Verify that the object was written to S3
        if event_name == 'ObjectCreated:____' and bucket_name == 'sd-vehicle-data':
           
            # If so, then add it to the written_objects list
            written_objects.append(obj_key)
    
    # Return the collected written objects
    return {
        "bucket": 'sd-vehicle-data',
        "new_objects": written_objects
    }

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json


# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("recordReaderS3", "record_created_handler")

# Read the test event JSON file as dictionary
with open("./dcs/sample_events/s3_event_c2l1.json") as json_file:
    payload = json.load(json_file)

# Invoke lambda with test event, print result
result = invoke_function_and_get_message("recordReaderS3", payload)
print(json.dumps(json.loads(result), indent=4))

# EDIT ME!
# recordReaderS3/lambda_function.py
# This is lambda handler code that will execute each time an object is added to S3
import json

def record_created_handler(event, context):
    # Create a list to store new object keys
    written_objects = []

    # Iterate over each S3 event record
    for record in event['Records']:
       
        # Get the variables to check for
        event_name = record['eventName']
        bucket_name = record['s3']['bucket']['name']
        obj_key = record['s3']['object']['key']

        # Verify that the object was written to S3
        if event_name == 'ObjectCreated:Put' and bucket_name == 'sd-vehicle-data':
           
            # If so, then add it to the written_objects list
            written_objects.append(obj_key)
    
    # Return the collected written objects
    return {
        "bucket": 'sd-vehicle-data',
        "new_objects": written_objects
    }                 

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event
from dcs.util.aws_lambda import * 
import json


# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("recordReaderS3", "record_created_handler")

# Read the test event JSON file as dictionary
with open("./dcs/sample_events/s3_event_c2l1.json") as json_file:
    payload = json.load(json_file)

# Invoke lambda with test event, print result
result = invoke_function_and_get_message("recordReaderS3", payload)
print(json.dumps(json.loads(result), indent=4))

Lectura de datos de S3 basada en un evento

¡Tu primer lambda en directo!

Añadir una capa lambda

Lambda Flow

Lambda flow

Utilizar variables de entorno

Flujo de trabajo de datos sin servidor

Lambdas de ejecución rápida

Lambdas que tardan más en ejecutarse

Lambdas rápidas vs. lentas

Creación de lambda temporizada

Una lambda temporizada

Código lambda temporizado

Sin servidor APIs

Crear API-Lambda activada

Crea una lambda API

import boto3

lambda_client = boto3.client('lambda', aws_access_key_id="12345",
    aws_secret_access_key="12345",
    region_name='us-east-1',
    endpoint_url="http://localhost:4574")

def trigger_full_report_builder(event, context):
    async = event['____']['async']
    if async == 'true':
        exec_type = '____'
    else:
        exec_type = 'RequestResponse'
        
    result = lambda_client.____(
      FunctionName='fullReportBuilder',
      InvocationType=exec_type)
    
    return {
        'statusCode': 200,
        'headers': {
            "content-type" : "application/json"
        },
        'body': "Successfully Invoked"
    }

import awswrangler_c as wr
import boto3
# Initialize boto3 session
session = boto3.Session(aws_access_key_id = "12345", 
  aws_secret_access_key = "12345", 
  region_name="us-east-1")

def full_report_builder(event, context):
  # Read all records in the speeders folder
  speeders_total = wr.s3.read_csv('s3://sd-vehicle-data/speeders', 
                                  boto3_session = session, 
                                  delimiter=" ")

  # Write aggregated speeders file
  return wr.s3.to_csv(df = speeders_total, 
               boto3_session=session, 
               path="s3://sd-vehicle-data/speeders-full/full.csv")
               

# PLEASE DO NOT EDIT
# aws_wrangler_c.py
def read_csv(path, boto3_session, delimiter):
  bucket = path.split("//")[1].split("/")[0]
  prefix = path.split("//")[1].replace(f"{bucket}/", "")
  objects = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)['Contents']
  dfs = []
  for o in objects:
      key = o['Key']
      data = s3.get_object(Bucket='sd-vehicle-data', Key=key)
      dfs.append(pd.read_csv(data['Body'], delimiter=delimiter))
  return pd.concat(dfs)
  
def to_csv(df, path, boto3_session, separator=","):
  bucket = path.split("//")[1].split("/")[0]
  key = path.split("//")[1].replace(f"{bucket}/", "")
  s3.put_object(Bucket=bucket, Key=key, Body=df.to_csv(sep=separator, index=False))
  return True

s3 = {
    "to_csv": to_csv,
    "read_csv": read_csv
}
                                                  

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json

# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("fullReportBuilder", "full_report_builder")
create_lambda("triggerFullReportBuilder", "trigger_full_report_builder")


# Invoke lambda with test event, print result.
result = invoke_function_and_get_message("triggerFullReportBuilder", {
  "queryStringParameters": {
    "async": "true"
  },
  "isBase64Encoded": "false"
})
print(result)

import boto3

lambda_client = boto3.client('lambda', aws_access_key_id="12345",
    aws_secret_access_key="12345",
    region_name='us-east-1',
    endpoint_url="http://localhost:4574")

def trigger_full_report_builder(event, context):
    async = event['queryStringParameters']['async']
    if async == 'true':
        exec_type = 'Event'
    else:
        exec_type = 'RequestResponse'
        
    result = lambda_client.invoke(
      FunctionName='fullReportBuilder',
      InvocationType=exec_type)
    
    return {
        'statusCode': 200,
        'headers': {
            "content-type" : "application/json"
        },
        'body': "Successfully Invoked"
    }

    

import awswrangler_c as wr
import boto3
# Initialize boto3 session
session = boto3.Session(aws_access_key_id = "12345", 
  aws_secret_access_key = "12345", 
  region_name="us-east-1")

def full_report_builder(event, context):
  # Read all records in the speeders folder
  speeders_total = wr.s3.read_csv('s3://sd-vehicle-data/speeders', 
                                  boto3_session = session, 
                                  delimiter=" ")

  # Write aggregated speeders file
  return wr.s3.to_csv(df = speeders_total, 
               boto3_session=session, 
               path="s3://sd-vehicle-data/speeders-full/full.csv")
               
  

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json

# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("fullReportBuilder", "full_report_builder")
create_lambda("triggerFullReportBuilder", "trigger_full_report_builder")

# Invoke lambda with test event, print result.
result = invoke_function_and_get_message("triggerFullReportBuilder", {
  "queryStringParameters": {
    "async": "true"
  },
  "isBase64Encoded": "false"
})
print(result)

Un servidor sin API

Ya estás preparado para codificar y descodificar datos de flujo y analizarlos directamente en el flujo. Incluso utilizarás varios flujos para obtener las velocidades máximas diarias de los vehículos.

Una lambda transformadora

Transformar datos dentro de un flujo

Codificar y descodificar base64

Crea una lambda transformacional

Una lambda transformacional básica

Analizar datos en el flujo

Crear una aplicación de análisis de datos Kinesis

Análisis de datos Kinesis frente a Lambda transformacional

Construir una aplicación de análisis de datos Kinesis

Componentes de análisis de datos Kinesis SQL

Utilizar varios flujos

Entrega de datos de Kinesis Analytics

Consigue la velocidad máxima diaria

Utilizar el análisis de datos Kinesis para las alertas

Analizar datos de streaming

En este último capítulo, descubrirás cómo supervisar el rendimiento de tu flujo utilizando registros, métricas, alarmas y cuadros de mando. Utilizarás Elasticsearch y construirás tu propio panel de control Kibana.

Estudio de caso de streaming de datos

Crear un clúster de Elasticsearch

RedShift vs Elasticsearch

Control del rendimiento

Cloudwatch dasbhoards y alarmas

Componentes de Cloudwatch

Flujo de monitorización de Cloudwatch

Visualizar datos en streaming

Trabajar con ElasticSearch utilizando Kibana

Cloudwatch vs Elasticsearch

Flujo de visualización de Elasticsearch

Un enfoque alternativo

Otro enfoque alternativo

Monitorización y visualización de datos en streaming

¿Qué hace funcionar los sistemas que utilizamos todos los días sin darnos cuenta, como la detección de fraudes para mantener seguras nuestras transacciones o las señales de tráfico que mantienen la fluidez del tráfico? La respuesta son los datos en streaming (datos que se generan continuamente a partir de distintas fuentes) y las tecnologías sin servidor, como Amazon Kinesis y AWS Lambda.
<br /> <br />
En este curso, aprenderás a aprovechar estas potentes tecnologías ayudando a un ingeniero de datos ficticio llamado Cody. Tu objetivo es ayudarla a recopilar datos de flujo en tiempo real de los vehículos propiedad de la ciudad, analizar los datos y enviar alertas relevantes, como avisos de velocidad, a los conductores. Utilizando Amazon Kinesis y Firehose, aprenderás a ingerir datos de millones de fuentes antes de utilizar Kinesis Analytics para analizar los datos a medida que se mueven por el flujo. También crearás funciones sin servidor en AWS Lambda que desencadenarán condicionalmente acciones basadas en los datos recibidos. Al final de esta formación sabrás cómo crear cuadros de mando de ElasticSearch en directo con AWS QuickSight y CloudWatch, y con suerte habrás ayudado a Cody a completar su ambicioso proyecto.


Introduction to AWS Boto in Python

Introduction to Shell

Streaming Concepts

Aprende a trabajar con datos en streaming usando tecnologías serverless en AWS.

Streaming de datos con AWS Kinesis y Lambda

Aprende a trabajar con datos de streaming utilizando tecnologías sin servidor en AWS.

Data Engineer

Adding a lambda layer

Let's practice!

Sin servidores - Añadir una capa lambda

Añadir una capa lambda

Create Your Free Account