În acest prim capitol, vei afla care sunt diferențele dintre procesarea în lot și procesarea în flux, vei crea primul tău flux, îi vei gestiona permisiunile și vei scrie și citi date din el.

Ce este streaming-ul și de ce contează?

Opțiuni

Batch

Stream

Loturi vs. flux

Producători de date

...introduc date în fluxul Firehose

...care le livrează către Destinație

Producători și destinații

Gestionarea fluxurilor de livrare Firehose

Pregătire pentru primul flux

Crearea rolurilor

Utilizator IAM

Atât Utilizatori IAM, cât și Roluri

Utilizatori IAM vs. roluri IAM

Permisiuni doar pentru utilizator

Permisiuni pentru utilizator și rol

Exercițiu cu permisiuni

Crearea unui bucket S3

Lucrul cu fluxul de livrare 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']))


Creează primul tău 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']))


Scrierea într-un 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())


Citirea datelor din Firehose

Streaming în cloud

Următorul pas în parcursul tău cu date în flux este să înveți cum să folosești funcții lambda de transformare pentru a lucra serverless. Prin exerciții practice, vei adăuga straturi lambda și vei declanșa funcții lambda în condiții specifice.

Trecerea la serverless

Crearea și rularea funcțiilor Lambda

Serverless vs. servere

Componentele unei funcții Lambda

Citirea datelor din S3 pe baza unui eveniment

Prima ta funcție lambda live!

Adăugarea unui layer Lambda

Fluxul Lambda

Utilizarea variabilelor de mediu

Flux de date serverless

Lambda-uri rapide vs. lente

O funcție lambda declanșată la un moment fix

Cod lambda declanșat la interval de timp

API-uri serverless

Creează o funcție Lambda pentru API

Un API serverless

Acum ești pregătit să codezi și să decodezi date în flux și să le analizezi direct în flux. Vei folosi chiar mai multe fluxuri pentru a obține vitezele maxime zilnice ale vehiculelor.

Un Lambda transformațional

Transformarea datelor într-un flux

Codificare și decodificare base64

Creează o funcție Lambda de transformare

O funcție lambda de transformare minimală

Analizarea datelor în flux

Crearea unei aplicații de analiză a datelor Kinesis

Kinesis Data Analytics vs. Lambda transformațional

Crearea unei aplicații Kinesis Data Analytics

Componentele SQL din Kinesis Data Analytics

Utilizarea mai multor fluxuri

Livrarea datelor din Kinesis Analytics

Obține viteza maximă zilnică

Utilizarea Kinesis Data Analytics pentru alerte

Analiza datelor în flux

În acest capitol final, vei descoperi cum să monitorizezi performanța fluxului tău folosind jurnale, metrici, alarme și dashboarduri. Vei folosi Elasticsearch și îți vei construi propriul dashboard Kibana.

Studiu de caz: date în flux continuu

Crearea unui cluster Elasticsearch

RedShift vs Elasticsearch

Monitorizarea performanței

Dashboard-uri și alarme în CloudWatch

Componentele CloudWatch

Fluxul de monitorizare CloudWatch

Vizualizarea datelor în flux

Lucrul cu ElasticSearch folosind Kibana

CloudWatch vs. Elasticsearch

Fluxul de vizualizare în Elasticsearch

O abordare alternativă

O altă abordare alternativă

Monitorizarea și vizualizarea datelor în flux

Ce stă în spatele sistemelor pe care le folosim zilnic fără să ne dăm seama – de la detectarea fraudelor pentru a ne proteja tranzacțiile, până la semafoarele inteligente care fluidizează traficul? Răspunsul este datele în flux (date generate continuu din surse diverse) și tehnologiile serverless – precum Amazon Kinesis și AWS Lambda.
<br /> <br />
În acest curs, vei învăța cum să folosești aceste tehnologii puternice ajutând-o pe Cody, o ingineră de date fictivă. Obiectivul tău este să o ajuți să colecteze date în timp real de la vehiculele orașului, să le analizeze și să trimită alerte relevante – cum ar fi avertismente de viteză pentru șoferi. Folosind Amazon Kinesis și Firehose, vei învăța cum să preiei date din milioane de surse, apoi vei folosi Kinesis Analytics pentru a analiza datele pe măsură ce trec prin flux. Vei crea și funcții serverless în AWS Lambda care declanșează acțiuni în mod condiționat, în funcție de datele primite. La finalul cursului, vei ști cum să construiești dashboarduri live în ElasticSearch cu AWS QuickSight și CloudWatch – și, sperăm, o vei fi ajutat pe Cody să își ducă la bun sfârșit proiectul ambițios.


Introduction to AWS Boto in Python

Introduction to Shell

Streaming Concepts

Învață să lucrezi cu date de streaming folosind tehnologii serverless pe AWS.

Date în flux cu AWS Kinesis și Lambda

Învață să lucrezi cu date în streaming folosind tehnologii serverless pe AWS.

Ce este streaming-ul și de ce contează?

Create Your Free Account