In this first chapter, you will learn about the differences between batch and stream processing, create your first stream, manage its permissions, write to it and read from it.

Che cos’è lo streaming e perché è importante?

Opzioni

Batch

Stream

Batch vs stream

Produttori di dati

..inseriscono i dati nel Firehose Stream

...che li consegna alla Destinazione

Produttori e destinazioni

Gestire i delivery stream di Firehose

Prepararsi al primo stream

Creare ruoli

IAM User

Role

Sia IAM Users che Roles

Utenti IAM vs. ruoli

Permissions for User only

Permissions for User and Role

Esercizio sulle autorizzazioni

Creazione di un bucket S3

Lavorare con il delivery stream di 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 il tuo primo stream 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']))


Scrivere in uno stream 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())


Lettura dei dati Firehose

Streaming in the cloud

The next step in your streaming data journey is learning how to use transformational lambda functions to go serverless. Through hands-on exercises, you will add lambda layers and trigger lambda functions on specific conditions.

Going serverless

Creating and running Lambda functions

Serverless vs servers

Lambda function components

Reading data from S3 based on an event

Your first live lambda!

Adding a lambda layer

Lambda flow

Using environment variables

Serverless data workflow

Fast vs slow lambdas

A time-triggered lambda

Time-triggered lambda code

Serverless APIs

Create an API lambda

A serverless API

You're now ready to encode and decode streaming data and analyze data directly in the stream. You will even use multiple streams to get daily vehicle top speeds.

A transformational Lambda

Transforming data inside a stream

Encoding and decoding base64

Create a transformational lambda

A barebones transformational lambda

Analyzing data in the stream

Creating a Kinesis data analytics application

Kinesis data analytics vs transformational Lambda

Building a Kinesis data analytics application

Kinesis data analytics SQL components

Using multiple streams

Delivering data from Kinesis Analytics

Get the daily top speed

Using Kinesis data analytics for alerts

Analyzing streaming data

In this final chapter, you will discover how to monitor your stream's performance using logs, metrics, alarms and dashboards. You will use Elasticsearch and build your own Kibana dashboard.

Streaming data case study

Creating an Elasticsearch cluster

RedShift vs Elasticsearch

Monitoring performance

Cloudwatch dasbhoards and alarms

Cloudwatch components

Cloudwatch monitoring flow

Visualizing streaming data

Working with ElasticSearch using Kibana

Cloudwatch vs Elasticsearch

Elasticsearch visualization flow

An alternative approach

Another alternative approach

Monitoring and visualizing streaming data

Cosa alimenta, senza che ce ne accorgiamo, i sistemi che usiamo ogni giorno, come il rilevamento delle frodi che protegge le nostre transazioni o i semafori che tengono fluido il traffico? La risposta è lo streaming dei dati (dati generati in modo continuo da fonti diverse) e le tecnologie serverless—come Amazon Kinesis e AWS Lambda.
<br /> <br />
In questo corso imparerai a sfruttare queste tecnologie potenti aiutando una data engineer fittizia di nome Cody. Il tuo obiettivo è aiutarla a raccogliere dati in streaming in tempo reale dai veicoli di proprietà della città, analizzarli e inviare avvisi pertinenti, come segnalazioni di velocità, ai conducenti. Con Amazon Kinesis e Firehose imparerai a ingerire dati da milioni di fonti, per poi usare Kinesis Analytics per analizzare i dati mentre scorrono nello stream. Attiverai anche funzioni serverless in AWS Lambda che eseguiranno azioni in modo condizionale in base ai dati ricevuti. Al termine del corso saprai creare dashboard ElasticSearch in tempo reale con AWS QuickSight e CloudWatch—e, si spera, avrai aiutato Cody a portare a termine il suo ambizioso progetto.


Introduction to AWS Boto in Python

Introduction to Shell

Streaming Concepts

Scopri come gestire dati in streaming con tecnologie serverless su AWS.

Dati in streaming con AWS Kinesis e Lambda

Scopri come lavorare con i dati in streaming usando le tecnologie serverless su AWS.

Dispositivo di misurazione collegato al veicolo.

Produttori e destinazioni

Dati in streaming con AWS Kinesis e Lambda

Esercizio pratico interattivo