使用 Kinesis 擷取串流資料

PDFRSS

此程序會示範如何從名為 ev_station_data 的 Kinesis 串流中擷取資料，該串流包含來自不同 EV 充電站的取用資料 (採用 JSON 格式)。該結構描述是明確定義的。此範例會示範如何將資料儲存為原始 JSON，以及如何在擷取資料時將 JSON 資料轉換為 Amazon Redshift 資料類型。

生產者設定

1. 使用 Amazon Kinesis Data Streams，按照以下步驟建立名為 ev_station_data 的串流。選擇隨需做為容量模式。如需詳細資訊，請參閱透過 AWS 管理主控台建立串流。

2. Amazon Kinesis 資料產生器可協助您產生用於串流的測試資料。請依照工具中的詳細步驟來開始使用，並使用下列資料範本來產生資料：

   {
       
      "_id" : "{{random.uuid}}",
      "clusterID": "{{random.number(
           {   "min":1,
               "max":50
           }
       )}}", 
       "connectionTime": "{{date.now("YYYY-MM-DD HH:mm:ss")}}",
       "kWhDelivered": "{{commerce.price}}",
       "stationID": "{{random.number(
           {   "min":1,
               "max":467
           }
       )}}",
         "spaceID": "{{random.word}}-{{random.number(
           {   "min":1,
               "max":20
           }
       )}}",
    
      "timezone": "America/Los_Angeles",
      "userID": "{{random.number(
           {   "min":1000,
               "max":500000
           }
       )}}"
   }

   串流資料中的每個 JSON 物件都包含下列屬性：

   {
       "_id": "12084f2f-fc41-41fb-a218-8cc1ac6146eb",
       "clusterID": "49",
       "connectionTime": "2022-01-31 13:17:15",
       "kWhDelivered": "74.00",
       "stationID": "421",
       "spaceID": "technologies-2",
       "timezone": "America/Los_Angeles",
       "userID": "482329"
   }

Amazon Redshift 設定

這些步驟會說明如何設定具體化視觀表以擷取資料。

1. 建立外部結構描述以將 Kinesis 中的資料對應至 Redshift 物件。

   CREATE EXTERNAL SCHEMA evdata FROM KINESIS
   IAM_ROLE 'arn:aws:iam::0123456789:role/redshift-streaming-role';

   如需如何設定 IAM 角色的詳細資訊，請參閱 開始使用 Amazon Kinesis Data Streams 中的串流擷取。

2. 建立具體化視觀表以取用串流資料。下列範例示範如何定義具體化檢視，以從 Kinesis 串流擷取 JSON 格式的資料。

   首先，以半結構化 SUPER 格式儲存串流記錄。在此範例中，JSON 來源會儲存在 Redshift 中，而不會轉換為 Redshift 類型。

   CREATE MATERIALIZED VIEW ev_station_data AS
       SELECT approximate_arrival_timestamp,
       partition_key,
       shard_id,
       sequence_number,
       case when can_json_parse(kinesis_data) then json_parse(kinesis_data) else null end as payload,
       case when not can_json_parse(kinesis_data) then kinesis_data else null end as failed_payload
       FROM evdata."ev_station_data" ;

查詢串流

1. 使用下列命令啟用區分大小寫的 SUPER 屬性。Amazon Redshift 預設不區分大小寫，因此若要存取區分大小寫的 SUPER 屬性，您需要啟用此功能。

   SET enable_case_sensitive_super_attribute to TRUE;

2. 使用以下命令重新整理具體化檢視，以便從串流提取資料。

   REFRESH MATERIALIZED VIEW ev_station_data;

3. 查詢重新整理的具體化視觀表以取得使用狀況統計資料。

   SELECT e.payload.connectionTime::date as connectiontime
   ,SUM(e.payload.kWhDelivered::decimal(10,2)) AS Energy_Consumed
   ,count(distinct e.payload.userID) AS #Users
   from ev_station_data as e
   group by connectiontime
   order by 1 desc;

4. 檢視結果。

   connectiontime        energy_consumed    #users
   2022-02-08                 4139          10
   2022-02-09                 5571          10
   2022-02-10                 8697          20
   2022-02-11                 4408          10
   2022-02-12                 4257          10
   2022-02-23                 6861          10