在第一章，你會學到批次處理與串流處理的差異，建立你的第一個串流、管理其權限，並對該串流寫入與讀取資料。

什麼是串流處理？為什麼重要？

選項

Batch

Stream

批次 vs 串流

Data Producers

..把資料放進 Firehose Stream

...接著將資料送到 Destination

Producers 與 Destinations

管理 Firehose 傳送串流

準備建立第一個串流

建立角色

IAM 使用者

角色

IAM 使用者與角色皆可

IAM 使用者 vs. 角色

僅限使用者的權限（Permissions for User only）

使用者與 Role 的權限（Permissions for User and Role）

權限練習

建立 S3 Bucket

使用 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")


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


建立你的第一個 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")
    

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


寫入 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")
    

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 資料

在雲端進行串流

在你的串流資料旅程中，下一步是學會使用具轉換功能的 Lambda 函式來邁向無伺服器。透過動手做的練習，你會新增 Lambda Layers，並在特定條件下觸發 Lambda 函式。

走向無伺服器

建立並執行 Lambda 函式

Serverless 與伺服器

Lambda 函式元件

根據事件從 S3 讀取資料

你的第一個即時 Lambda！

新增 Lambda Layer

Lambda 流程

使用環境變數

無伺服器資料工作流程

快速 vs 緩慢的 lambdas

時間觸發的 Lambda

時間觸發的 Lambda 程式碼

無伺服器 API

建立 API Lambda

邁向無伺服器

你現在已準備好對串流資料進行編碼與解碼，並直接在串流中分析資料。你甚至會使用多個串流來取得每日車輛的最高車速。

一個具有轉換功能的 Lambda

在資料串流中轉換資料

編碼與解碼 base64

建立轉換用的 Lambda

簡易版轉換型 lambda

在資料串流中進行分析

建立 Kinesis 資料分析應用程式

Kinesis Data Analytics 與轉換型 Lambda

建立 Kinesis Data Analytics 應用程式

Kinesis Data Analytics 的 SQL 元件

使用多個串流

從 Kinesis Analytics 傳送資料

取得每日最高車速

使用 Kinesis Data Analytics 發送警示

分析串流資料

在最後一章，你將學會如何利用日誌、度量、警報與儀表板來監控串流的效能。你會使用 Elasticsearch，並建立自己的 Kibana 儀表板。

串流資料個案研究

建立 Elasticsearch 叢集

RedShift 與 Elasticsearch

監控效能

CloudWatch 儀表板與警示

CloudWatch 元件

CloudWatch 監控流程

視覺化串流資料

使用 Kibana 操作 ElasticSearch

CloudWatch 與 Elasticsearch

Elasticsearch 視覺化流程

另一種做法

另一種替代做法

監控與視覺化串流資料

是什麼在幕後驅動我們日常不經意就使用的系統，例如保護交易安全的詐欺偵測，或讓車流順暢的號誌控制？答案是串流資料（由不同來源持續產生的資料）與無伺服器技術——例如 Amazon Kinesis 與 AWS Lambda。
<br /> <br />
在這門課中，你將透過協助一位名叫 Cody 的虛構資料工程師，學會如何運用這些強大的技術。你的目標是協助她從市有車輛蒐集即時串流資料、分析資料，並將相關警示（如超速警告）發送給駕駛。你會使用 Amazon Kinesis 與 Firehose，學習如何從成千上萬個來源擷取資料，接著使用 Kinesis Analytics 在資料於串流中流動時即時分析。此外，你也會在 AWS Lambda 中建立無伺服器函式，根據接收到的資料條件式觸發動作。完成本訓練後，你將知道如何使用 AWS QuickSight 與 CloudWatch 建立即時的 ElasticSearch 儀表板——也希望你已協助 Cody 完成她雄心勃勃的專案。


什麼是串流處理？為什麼重要？

Create Your Free Account