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

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

批次 vs 串流

Producers 與 Destinations

管理 Firehose 傳送串流

準備建立第一個串流

建立角色

IAM 使用者 vs. 角色

權限練習

建立 S3 Bucket

使用 Firehose 傳送串流

建立你的第一個 Firehose 串流

寫入 Firehose 串流

讀取 Firehose 資料

在雲端進行串流

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

走向無伺服器

建立並執行 Lambda 函式

選項

適合伺服器

適合 Serverless

Serverless 與伺服器

Trigger

Handler

Layer

Lambda 函式元件

# EDIT ME!
# recordReaderS3/lambda_function.py
# This is lambda handler code that will execute each time an object is added to S3.
import json

def record_created_handler(____, ____):
    # Create a list to store new object keys
    written_objects = []

    # Iterate over each S3 event record
    for record in event['Records']:
       
        # Get the variables to check for
        event_name = record['eventName']
        bucket_name = record['s3']['bucket']['name']
        obj_key = record['s3']['object']['key']

        # Verify that the object was written to S3
        if event_name == 'ObjectCreated:____' and bucket_name == 'sd-vehicle-data':
           
            # If so, then add it to the written_objects list
            written_objects.append(obj_key)
    
    # Return the collected written objects
    return {
        "bucket": 'sd-vehicle-data',
        "new_objects": written_objects
    }

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json


# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("recordReaderS3", "record_created_handler")

# Read the test event JSON file as dictionary
with open("./dcs/sample_events/s3_event_c2l1.json") as json_file:
    payload = json.load(json_file)

# Invoke lambda with test event, print result
result = invoke_function_and_get_message("recordReaderS3", payload)
print(json.dumps(json.loads(result), indent=4))

# EDIT ME!
# recordReaderS3/lambda_function.py
# This is lambda handler code that will execute each time an object is added to S3.
import json

def record_created_handler(event, context):
    # Create a list to store new object keys
    written_objects = []

    # Iterate over each S3 event record
    for record in event['Records']:
       
        # Get the variables to check for
        event_name = record['eventName']
        bucket_name = record['s3']['bucket']['name']
        obj_key = record['s3']['object']['key']

        # Verify that the object was written to S3
        if event_name == 'ObjectCreated:Put' and bucket_name == 'sd-vehicle-data':
           
            # If so, then add it to the written_objects list
            written_objects.append(obj_key)
    
    # Return the collected written objects
    return {
        "bucket": 'sd-vehicle-data',
        "new_objects": written_objects
    }

根據事件從 S3 讀取資料

你的第一個即時 Lambda！

新增 Lambda Layer

Lambda 流程

使用環境變數

無伺服器資料工作流程

執行較快的 Lambdas

執行時間較長的 Lambdas

快速 vs 緩慢的 lambdas

建立時間觸發的 Lambda

時間觸發的 Lambda

時間觸發的 Lambda 程式碼

無伺服器 API

建立由 API 觸發的 Lambda

建立 API Lambda

import boto3

lambda_client = boto3.client('lambda', aws_access_key_id="12345",
    aws_secret_access_key="12345",
    region_name='us-east-1',
    endpoint_url="http://localhost:4574")

def trigger_full_report_builder(event, context):
    is_async = event['____']['async']
    if is_async == 'true':
        exec_type = '____'
    else:
        exec_type = 'RequestResponse'
        
    result = lambda_client.____(
      FunctionName='fullReportBuilder',
      InvocationType=exec_type)
    
    return {
        'statusCode': 200,
        'headers': {
            "content-type" : "application/json"
        },
        'body': "Successfully Invoked"
    }

import awswrangler_c as wr
import boto3
# Initialize boto3 session
session = boto3.Session(aws_access_key_id = "12345", 
  aws_secret_access_key = "12345", 
  region_name="us-east-1")

def full_report_builder(event, context):
  # Read all records in the speeders folder
  speeders_total = wr.s3.read_csv('s3://sd-vehicle-data/speeders', 
                                  boto3_session = session, 
                                  delimiter=" ")

  # Write aggregated speeders file
  return wr.s3.to_csv(df = speeders_total, 
               boto3_session=session, 
               path="s3://sd-vehicle-data/speeders-full/full.csv")
               

# PLEASE DO NOT EDIT
# aws_wrangler_c.py
def read_csv(path, boto3_session, delimiter):
  bucket = path.split("//")[1].split("/")[0]
  prefix = path.split("//")[1].replace(f"{bucket}/", "")
  objects = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)['Contents']
  dfs = []
  for o in objects:
      key = o['Key']
      data = s3.get_object(Bucket='sd-vehicle-data', Key=key)
      dfs.append(pd.read_csv(data['Body'], delimiter=delimiter))
  return pd.concat(dfs)
  
def to_csv(df, path, boto3_session, separator=","):
  bucket = path.split("//")[1].split("/")[0]
  key = path.split("//")[1].replace(f"{bucket}/", "")
  s3.put_object(Bucket=bucket, Key=key, Body=df.to_csv(sep=separator, index=False))
  return True

s3 = {
    "to_csv": to_csv,
    "read_csv": read_csv
}
                                                  

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json

# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("fullReportBuilder", "full_report_builder")
create_lambda("triggerFullReportBuilder", "trigger_full_report_builder")


# Invoke lambda with test event, print result.
result = invoke_function_and_get_message("triggerFullReportBuilder", {
  "queryStringParameters": {
    "async": "true"
  },
  "isBase64Encoded": "false"
})
print(result)

import boto3

lambda_client = boto3.client('lambda', aws_access_key_id="12345",
    aws_secret_access_key="12345",
    region_name='us-east-1',
    endpoint_url="http://localhost:4574")

def trigger_full_report_builder(event, context):
    is_async = event['queryStringParameters']['async']
    if is_async == 'true':
        exec_type = 'Event'
    else:
        exec_type = 'RequestResponse'
        
    result = lambda_client.invoke(
      FunctionName='fullReportBuilder',
      InvocationType=exec_type)
    
    return {
        'statusCode': 200,
        'headers': {
            "content-type" : "application/json"
        },
        'body': "Successfully Invoked"
    }

邁向無伺服器

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

一個具有轉換功能的 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 完成她雄心勃勃的專案。


根據事件從 S3 讀取資料

使用 AWS Kinesis 與 Lambda 的串流資料

動手互動練習