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?

Opções

Processamento em lote

Processamento em fluxo

Lote vs. fluxo

Produtores de dados

... colocar dados no Firehose Stream

...que o entrega ao Destino

Produtores e destinos

Gerenciar fluxos de fornecimento do Firehose

Preparando-se para a primeira transmissão

Criação de funções

IAM Usuário

Função

IAM Usuários e funções

IAM usuários vs. funções

Permissões somente para o usuário

Permissões para usuário e função

Prática de permissões

Criação de bucket S3

Trabalhando com o fluxo de entrega do 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']))


Crie seu primeiro fluxo do 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']))


Gravação em um fluxo do 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())


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

Sem servidor vs. servidores

Componentes da função lambda

Leitura de dados do S3 com base em um evento

Seu primeiro lambda ao vivo!

Adição de uma camada lambda

Fluxo lambda

Uso de variáveis de ambiente

Fluxo de trabalho de dados sem servidor

Lambdas rápidos versus lambdas lentos

Um lambda acionado por tempo

Código lambda acionado por tempo

Sem servidor APIs

Criar um lambda API

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.