I det här första kapitlet lär du dig skillnaderna mellan batchbearbetning och strömbearbetning. Du skapar din första ström, hanterar dess behörigheter och skriver till och läser från den.

Vad är streaming och varför är det viktigt?

Alternativ

Batch

Ström

Batch vs. ström

Dataproducenter

...skickar data till Firehose-strömmen

...som levererar det till destination

Producenter och destinationer

Hantera Firehose delivery streams

Förbered dig för den första streamen

Skapa roller

IAM-användare

Roll

Både IAM-användare och roller

IAM-användare vs. roller

Behörigheter endast för användare

Behörigheter för användare och roll

Behörighetsövning

Skapa en S3-bucket

Arbeta med Firehose-leveransströmmen

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


Skapa ditt första Firehose-flöde

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


Skriva till en Firehose-ström

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


Läsa Firehose-data

Strömning i molnet

Nästa steg på din resa med strömdata är att lära dig använda transformerande lambdafunktioner för att arbeta serverlöst. Genom praktiska övningar lägger du till lambda-lager och utlöser lambdafunktioner under specifika villkor.

Dags att gå serverlöst

Skapa och kör Lambda-funktioner

Serverlöst vs servrar

Komponenter i en lambdafunktion

Läsa data från S3 baserat på en händelse

Din första live-lambda!

Lägga till ett lambda-lager

Lambda-flödet

Använda miljövariabler

Serverlöst dataarbetsflöde

Snabba och långsamma lambdafunktioner

En tidsutlöst Lambda-funktion

Tidsutlöst lambda-kod

Serverlösa API:er

Skapa en API-lambda

Ett serverlöst API

Serverlös arkitektur

Nu är du redo att koda och avkoda strömdata samt analysera data direkt i strömmen. Du använder även flera strömmar för att ta fram fordonens dagliga topphastigheter.

En transformerande Lambda

Transformera data i en ström

Koda och avkoda base64

Skapa en transformerande Lambda-funktion

En enkel transformations-lambda

Analysera data i strömmen

Skapa en Kinesis-applikation för dataanalys

Kinesis Data Analytics vs transformations-Lambda

Bygg en Kinesis Data Analytics-applikation

SQL-komponenter i Kinesis Data Analytics

Använda flera strömmar

Leverera data från Kinesis Analytics

Hämta dagens topphastighet

Använda Kinesis Data Analytics för aviseringar

Analysera strömdata

I det här sista kapitlet får du lära dig hur du övervakar strömmens prestanda med hjälp av loggar, mätvärden, larm och dashboards. Du använder Elasticsearch och bygger din egen Kibana-dashboard.

Fallstudie: strömmande data

Skapa ett Elasticsearch-kluster

RedShift vs Elasticsearch

Övervaka prestanda

CloudWatch-instrumentpaneler och larm

CloudWatch-komponenter

Övervakningsflöde i CloudWatch

Visualisering av strömmande data

Arbeta med ElasticSearch via Kibana

CloudWatch vs Elasticsearch

Visualiseringsflöde i Elasticsearch

Ett alternativt tillvägagångssätt

Ett annat alternativt tillvägagångssätt

Övervaka och visualisera strömdata

Vad driver de system vi använder varje dag utan att tänka på det – som bedrägeriskydd för att hålla våra transaktioner säkra, eller trafiksignaler som får trafiken att flyta smidigt? Svaret är strömdata (data som kontinuerligt genereras från olika källor) och serverlös teknik – som Amazon Kinesis och AWS Lambda.
<br /> <br />
I den här kursen får du lära dig att använda dessa kraftfulla tekniker genom att hjälpa en fiktiv dataingenjör som heter Cody. Målet är att hjälpa henne att samla in realtidsströmdata från kommunala fordon, analysera datan och skicka relevanta varningar – till exempel hastighetsvarningar till förare. Med Amazon Kinesis och Firehose lär du dig att ta emot data från miljontals källor, och sedan använda Kinesis Analytics för att analysera data medan den rör sig genom strömmen. Du skapar även serverlösa funktioner i AWS Lambda som villkorsstyrt utlöser åtgärder baserat på inkommande data. I slutet av kursen vet du hur man skapar live-dashboards i ElasticSearch med AWS QuickSight och CloudWatch – och förhoppningsvis har du också hjälpt Cody att slutföra sitt ambitiösa projekt.


Vad är streaming och varför är det viktigt?

Create Your Free Account