Neste primeiro capítulo, você aprenderá sobre as diferenças entre processamento em lote e em fluxo, criará seu primeiro fluxo, gerenciará suas permissões, gravará nele e lerá a partir dele.

O que é streaming e por que ele é importante?

Lote vs. fluxo

Produtores e destinos

Gerenciar fluxos de fornecimento do Firehose

Preparando-se para a primeira transmissão

Criação de funções

IAM usuários vs. funções

Prática de permissões

Criação de bucket S3

Trabalhando com o fluxo de entrega do Firehose

Crie seu primeiro fluxo do Firehose

Gravação em um fluxo do Firehose

Leitura de dados do Firehose

Streaming na nuvem

A próxima etapa da sua jornada de dados de streaming é aprender a usar funções lambda transformacionais para ficar sem servidor. Por meio de exercícios práticos, você adicionará camadas lambda e acionará funções lambda em condições específicas.

Tornando-se sem servidor

Criar e executar funções Lambda

Opções

Ótimo para um servidor

Excelente para Serverless

Sem servidor vs. servidores

Gatilho

Manipulador

Camada

Componentes da função 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))

Leitura de dados do S3 com base em um evento

Seu primeiro lambda ao vivo!

Adição de uma camada lambda

Fluxo Lambda

Fluxo lambda

Uso de variáveis de ambiente

Fluxo de trabalho de dados sem servidor

Lambdas de execução rápida

Lambdas que demoram mais para serem executados

Lambdas rápidos versus lambdas lentos

Criando Lambda acionado por tempo

Um lambda acionado por tempo

Código lambda acionado por tempo

Sem servidor APIs

Criar API-Triggered Lambda

Criar um 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)

Um servidor sem servidor API

Agora você está pronto para codificar e decodificar dados de streaming e analisar dados diretamente no streaming. Você ainda usará vários fluxos para obter as velocidades máximas diárias dos veículos.

Um Lambda transformacional

Transformação de dados em um fluxo

Codificação e decodificação de base64

Criar um lambda transformacional

Um lambda transformacional básico

Análise de dados no fluxo

Criando um aplicativo de análise de dados do Kinesis

Análise de dados do Kinesis versus Lambda transformacional

Criação de um aplicativo de análise de dados do Kinesis

Análise de dados do Kinesis SQL componentes

Uso de vários fluxos

Fornecimento de dados do Kinesis Analytics

Obtenha a velocidade máxima diária

Usando a análise de dados do Kinesis para alertas

Análise de dados de fluxo contínuo

Neste capítulo final, você descobrirá como monitorar o desempenho do seu fluxo usando logs, métricas, alarmes e painéis. Você usará o Elasticsearch e criará seu próprio painel do Kibana.

Estudo de caso de dados em fluxo contínuo

Criando um cluster do Elasticsearch

RedShift vs Elasticsearch

Monitoramento do desempenho

Alarmes e dasbhoards do Cloudwatch

Componentes do Cloudwatch

Fluxo de monitoramento do Cloudwatch

Visualização de dados de fluxo contínuo

Trabalhando com o ElasticSearch usando o Kibana

Cloudwatch vs Elasticsearch

Fluxo de visualização do Elasticsearch

Uma abordagem alternativa

Outra abordagem alternativa

Monitoramento e visualização de dados de streaming

O que alimenta os sistemas que usamos todos os dias sem perceber, como a detecção de fraudes para manter nossas transações seguras ou os sinais de trânsito que mantêm o tráfego fluindo sem problemas? A resposta são os dados de streaming (dados gerados continuamente por diferentes fontes) e as tecnologias sem servidor, como o Amazon Kinesis e o AWS Lambda.
<br /> <br />
Neste curso, você aprenderá a aproveitar essas tecnologias poderosas ajudando um engenheiro de dados fictício chamado Cody. Seu objetivo é ajudá-la a coletar dados de streaming em tempo real de veículos de propriedade da cidade, analisar os dados e enviar alertas relevantes, como avisos de velocidade aos motoristas. Usando o Amazon Kinesis e o Firehose, você aprenderá a ingerir dados de milhões de fontes antes de usar o Kinesis Analytics para analisar os dados à medida que eles passam pelo fluxo. Você também criará funções sem servidor no AWS Lambda que acionarão condicionalmente ações com base nos dados recebidos. Ao final deste treinamento, você saberá como criar dashboards do ElasticSearch ao vivo com o AWS QuickSight e o CloudWatch e, com sorte, ajudará Cody a concluir seu ambicioso projeto.


Introduction to AWS Boto in Python

Introduction to Shell

Streaming Concepts

Aprenda a trabalhar com dados em streaming usando tecnologias serverless na AWS.

Streaming de dados com AWS Kinesis e Lambda

Saiba como trabalhar com dados de streaming usando tecnologias sem servidor no AWS.