In dit eerste hoofdstuk leer je het verschil tussen batch- en streamverwerking kennen, maak je je eerste stream, beheer je de rechten, en schrijf en lees je naar en van de stream.

Wat is streaming en waarom is het belangrijk?

Opties

Batch

Stream

Batch vs. stream

Options

Data Producers

..zetten data in de Firehose-stream

...die het levert aan Destination

Producers en destinations

Firehose-delivery streams beheren

Klaarmaken voor de eerste stream

Rollen maken

IAM-gebruiker

Zowel IAM-gebruikers als rollen

IAM-gebruikers versus rollen

Permissions for User only

Permissions for User and Role

Oefenen met machtigingen

S3-bucket maken

Werken met de Firehose-deliverystream

# _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")


from localstack_helpers import ensure_stream_absent

ensure_stream_absent(firehose, 'gps-delivery-stream')

# 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']))


# 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']))


Maak je eerste Firehose-stream

# _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")
    

from localstack_helpers import ensure_stream_absent
ensure_stream_absent(firehose, 'gps-delivery-stream')
# 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']))


Schrijven naar een Firehose-stream

# _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")
    

from localstack_helpers import ensure_stream_absent
ensure_stream_absent(firehose, 'gps-delivery-stream')
# 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())

# 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.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())

Firehose-gegevens lezen

Streamen in de cloud

De volgende stap in je streaming data-reis is leren hoe je transformerende lambda-functies gebruikt om serverless te gaan. In praktische oefeningen voeg je lambda-lagen toe en trigger je lambda-functies bij specifieke voorwaarden.

Serverless gaan

Lambda-functies maken en uitvoeren

Serverless vs servers

Onderdelen van een Lambda-functie

Gegevens uit S3 lezen op basis van een event

Je eerste live lambda!

Een lambda-laag toevoegen

Lambda-flow

Omgevingsvariabelen gebruiken

Serverloze dataworkflow

Snelle vs. langzame Lambdas

Een tijdgetriggerde Lambda

Tijdgestuurde lambda-code

Serverless API's

Maak een API-lambda

Een serverloze API

Je bent nu klaar om streaming data te encoderen en decoderen en data direct in de stream te analyseren. Je gebruikt zelfs meerdere streams om dagelijkse topsnelheden van voertuigen te krijgen.

Een transformerende Lambda

Data transformeren binnen een stream

Coderen en decoderen van base64

Maak een transformationele Lambda

Een minimalistische transformationele lambda

Data analyseren in de stream

Een Kinesis Data Analytics-toepassing maken

Kinesis Data Analytics vs transformationele Lambda

Een Kinesis Data Analytics-toepassing bouwen

Kinesis Data Analytics SQL-componenten

Meerdere streams gebruiken

Data leveren vanuit Kinesis Analytics

Bepaal de dagelijkse topsnelheid

Kinesis Data Analytics gebruiken voor meldingen

Streaming data analyseren

In dit laatste hoofdstuk ontdek je hoe je de prestaties van je stream monitort met logs, metrische gegevens, alarmen en dashboards. Je gebruikt Elasticsearch en bouwt je eigen Kibana-dashboard.

Praktijkcase streaming data

Een Elasticsearch-cluster maken

RedShift vs Elasticsearch

Prestaties monitoren

CloudWatch-dashboards en -alarmen

CloudWatch-onderdelen

CloudWatch-monitoringflow

Streaminggegevens visualiseren

Werken met ElasticSearch in Kibana

CloudWatch vs Elasticsearch

Stroom voor Elasticsearch-visualisaties

Een alternatieve aanpak

Nog een alternatieve aanpak

Streaming data monitoren en visualiseren

Wat drijft systemen die we dagelijks gebruiken zonder dat we het merken, zoals fraudedetectie die onze transacties beveiligt of verkeerslichten die het verkeer soepel laten doorstromen? Het antwoord is streaming data (data die continu door verschillende bronnen wordt gegenereerd) en serverless-technologieën—zoals Amazon Kinesis en AWS Lambda.
<br /> <br />
In deze cursus leer je hoe je deze krachtige technologieën inzet door een fictieve data-engineer, Cody, te helpen. Je doel is om haar te helpen realtime streaming data van gemeentelijke voertuigen te verzamelen, de data te analyseren en relevante waarschuwingen—zoals snelheidsmeldingen—naar bestuurders te sturen. Met Amazon Kinesis en Firehose leer je data in te lezen van miljoenen bronnen, waarna je Kinesis Analytics gebruikt om data te analyseren terwijl die door de stream beweegt. Je zet ook serverless-functies op in AWS Lambda die voorwaardelijk acties starten op basis van de ontvangen data. Aan het einde van deze training weet je hoe je live ElasticSearch-dashboards maakt met AWS QuickSight en CloudWatch—en heb je hopelijk Cody geholpen haar ambitieuze project te voltooien.


Introduction to AWS Boto in Python

Introduction to Shell

Streaming Concepts

Leer werken met streaming data via serverloze technologieën op AWS.

Streaming data met AWS Kinesis en Lambda

Leer werken met streaming data via serverless technologieën op AWS.

Meetinstrument aangesloten op het voertuig.

Producers en destinations

Streaming data met AWS Kinesis en Lambda

Interactieve oefening met praktijkervaring