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?

Opciones

Lote

Flujo

Lote vs flujo

Productores de datos

..poner los datos en el flujo Firehose

...que lo entrega en Destino

Productores y destinos

Gestionar flujos de entrega Firehose

Preparándose para el primer flujo

Crear roles

IAM Usuario

Función

Tanto IAM Usuarios y Roles

IAM usuarios vs. roles

Permisos sólo para Usuario

Permisos de Usuario y Rol

Práctica de permisos

Creación de bucket S3

Trabajar con el flujo de entrega Firehose

# _setup.py: Create clients, load data, etc. No need to edit.

import boto3

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")

# Prep variables for export
ex_vars = [firehose, s3]


# create_firehose.py: Create firehose stream. No need to edit.
import _setup
firehose, s3 = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.____(
    DeliveryStreamName="gps-delivery-stream",
    DeliveryStreamType="DirectPut",
    # Specify configuration of the destination S3 bucket
    ____ = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    
# Print the stream ARN
print("Created Firehose Stream ARN: {}".format(res['DeliveryStreamARN']))


# _setup.py: Create clients, load data, etc. No need to edit.
import boto3

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")

# Prep variables for export
ex_vars = [firehose, s3]


# create_firehose.py: Create firehose stream. No need to edit.
import _setup
firehose, s3 = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.create_delivery_stream(
    DeliveryStreamName =  "gps-delivery-stream",
    DeliveryStreamType = "DirectPut",
    
    # Specify configuration of the destination S3 bucket
    S3DestinationConfiguration = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    
# Print the stream ARN
print("Created Firehose Stream ARN: {}".format(res['DeliveryStreamARN']))


Crea tu primer flujo Firehose

# _setup.py: Create clients, load data, etc. No need to edit.
import boto3, pandas as pd

# Load Data
data_url = "https://assets.datacamp.com/production/repositories/5668/datasets/6bba555e0e42ae31d1d634256679db718cfb8d76/vehicles.csv"
records = pd.read_csv(data_url).head(100)

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")
    
# Prep variables for export
ex_vars = [firehose, s3, records]
    


# create_firehose.py: Create firehose stream. No need to edit.
import _setup
firehose, s3, records = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.create_delivery_stream(
    DeliveryStreamName =  "gps-delivery-stream",
    DeliveryStreamType = "DirectPut",
    
    # Specify configuration of the destination S3 bucket
    S3DestinationConfiguration = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    


# OBD2_sensors.py: Write to Firehose stream. EDIT HERE.
import _setup, create_firehose
firehose, s3, records = _setup.ex_vars
for idx, row in records.iterrows(): 

    # Create a payload string that ends with a newline
    payload = ' '.join(str(value) for value in row) 
    payload = payload + "\n"
    print("Sending payload: {}".format(payload))

    # Send the payload string to Firehose stream
    res = firehose.____(
        DeliveryStreamName = 'gps-delivery-stream',
        Record = {'____': payload})

    # Print the written RecordId
    print("Wrote to RecordId: {}".format(res['RecordId']))


# OBD2_sensors.py: Write to Firehose stream. EDIT HERE.
import _setup, create_firehose
firehose, s3, records = _setup.ex_vars
for idx, row in records.iterrows(): 

    # Create a payload string that ends with a newline
    payload = ' '.join(str(value) for value in row) 
    payload = payload + "\n"
    print("Sending payload: {}".format(payload))

    # Send the payload string to Firehose stream
    res = firehose.put_record(
        DeliveryStreamName = 'gps-delivery-stream',
        Record = {'Data': payload})

    # Print the written RecordId
    print("Wrote to RecordId: {}".format(res['RecordId']))


Escribir en un flujo Firehose

# _setup.py: Create clients, load data, etc. No need to edit.
import boto3, pandas as pd

# Load Data
data_url = "https://assets.datacamp.com/production/repositories/5668/datasets/6bba555e0e42ae31d1d634256679db718cfb8d76/vehicles.csv"
records = pd.read_csv(data_url).sample(100)

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")
    
# Prep variables for export
ex_vars = [firehose, s3, records]
    


# _run_deps.py: Execute dependencies. No need to edit.
import create_firehose
import OBD2_sensors


# create_firehose.py: Create firehose stream and s3 bucket. No need to edit.
import _setup
firehose, s3, records = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.create_delivery_stream(
    DeliveryStreamName =  "gps-delivery-stream",
    DeliveryStreamType = "DirectPut",
    
    # Specify configuration of the destination S3 bucket
    S3DestinationConfiguration = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    


# OBD2_sensors.py: Write OBD2 sensor data. No need to edit.
import _setup
firehose, s3, records = _setup.ex_vars
for idx, row in records.iterrows(): 

    # Create a payload string that ends with a newline
    payload = ' '.join(str(value) for value in row) 
    payload = payload + "\n"

    # Send the payload string to Firehose stream
    res = firehose.put_record(
        DeliveryStreamName = 'gps-delivery-stream',
        Record = {'Data': payload})
    


# analyze_data.py: Analyze written sensor data. EDIT HERE.
import _setup, _run_deps, pandas as pd
firehose, s3, records = _setup.ex_vars

# List the objects that have been written to the S3 bucket
objects = s3.____(Bucket='sd-vehicle-data')['Contents']

# Create list for collecting dataframes from read files.
dfs = []

# For every object, load it from S3
for obj in objects:
    data_file = s3.____(Bucket='sd-vehicle-data', ____=obj['Key'])

    # Load it into a dataframe, specifying a delimiter and column names
    dfs.append(pd.read_csv(data_file['Body'], 
                           delimiter = " ", 
                           names=["record_id", "timestamp", "vin", "lon", "lat", "speed"]))

# Concatenate the resulting dataframes.
data = pd.concat(dfs)
print(data.groupby(['vin'])['speed'].max())

# OBD2_sensors.py: Analyze written sensor data. EDIT HERE.
import _setup, _run_deps, pandas as pd
firehose, s3, records = _setup.ex_vars

# List the objects that have been written to the S3 bucket
objects = s3.list_objects(Bucket='sd-vehicle-data')['Contents']

# Create list for collecting dataframes from read files.
dfs = []

# For every object, load it from S3
for obj in objects:
    data_file = s3.get_object(Bucket='sd-vehicle-data', Key=obj['Key'])

    # Load it into a dataframe, specifying a delimiter and column names
    dfs.append(pd.read_csv(data_file['Body'], 
                           delimiter = " ", 
                           names=["record_id", "timestamp", "vin", "lon", "lat", "speed"]))

# Concatenate the resulting dataframes.
data = pd.concat(dfs)
print(data.groupby(['vin'])['speed'].max())


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

Sin servidor vs servidores

Componentes de la función lambda

Lectura de datos de S3 basada en un evento

¡Tu primer lambda en directo!

Añadir una capa lambda

Lambda flow

Utilizar variables de entorno

Flujo de trabajo de datos sin servidor

Lambdas rápidas vs. lentas

Una lambda temporizada

Código lambda temporizado

Sin servidor APIs

Crea una lambda API

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.