範例：使用 Regex (REGEX_REPLACE 函數) 替換子字串

PDFRSS

此範例使用 REGEX_REPLACE 函數來轉換 Amazon Kinesis Data Analytics 中的字串。 REGEX_REPLACE 用替代子字串替換一個子字串。如需詳細資訊，請參閱 Amazon Managed Service for Apache Flink SQL 參考資料中的REGEX_REPLACE。

在此範例中，將下列記錄寫入 Amazon Kinesis data stream：

{ "REFERRER" : "http://www.amazon.com" }
{ "REFERRER" : "http://www.amazon.com"}
{ "REFERRER" : "http://www.amazon.com"}
...

接著，您可以在主控台上建立 Kinesis Data Analytics 應用程式，並將 Kinesis 資料串流做為串流來源。探索程序會讀取串流來源上的範例記錄，並推斷含有一個資料欄 (REFERRER) 的應用程式內結構描述，如下所示。

然後，將應用程式碼與 REGEX_REPLACE 函數搭配使用，以轉換 URL 使用 https:// 而不是 http://。接著將產生的資料插入另一個應用程式內串流，如下所示：

步驟 1：建立 Kinesis 資料串流

建立 Amazon Kinesis 資料串流，並填入日誌紀錄，如下所示：

1. 登入 AWS Management Console 並開啟 Kinesis 主控台，網址為 https：//https://console.aws.amazon.com/kinesis。

2. 在導覽窗格中選擇資料串流。

3. 選擇建立 Kinesis 串流，然後建立內含一個碎片之串流。如需詳細資訊，請參閱 Amazon Kinesis Data Streams 開發人員指南中的建立串流。

4. 執行下列 Python 程式碼，以填入範例日誌記錄。這個簡單的代碼會持續寫入相同的日誌記錄到串流。

    
   import json
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   def get_data():
       return {"REFERRER": "http://www.amazon.com"}
   
   
   def generate(stream_name, kinesis_client):
       while True:
           data = get_data()
           print(data)
           kinesis_client.put_record(
               StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
           )
   
   
   if __name__ == "__main__":
       generate(STREAM_NAME, boto3.client("kinesis"))
   

步驟 2：建立 Kinesis Data Analytics 應用程式

接下來，建立 Kinesis Data Analytics 應用程式，如下所示：

1. 前往 https://console.aws.amazon.com/kinesisanalytics 開啟 Managed Service for Apache Flink 主控台。

2. 選擇建立應用程式，輸入應用程式名稱，然後選擇建立應用程式。

3. 在應用程式詳細資料頁面上，選擇連接串流資料。

4. 在連接至來源頁面，執行下列動作：

   1. 選擇您在上一節建立的串流。

   2. 選擇建立 IAM 角色 選項。

   3. 選擇探索結構描述。等待主控台顯示推斷的結構描述和範例記錄，這些記錄可用來推斷應用程式內串流所建立的結構描述。推斷的結構描述只有一個資料欄。

   4. 選擇儲存並繼續。

5. 在應用程式詳細資訊頁面上，選擇至 SQL 編輯器。若要啟動應用程式，請在出現的對話方塊中選擇是，啟動應用程式。

6. 在 SQL 編輯器中，編寫應用程式碼並驗證結果，如下所示：

   1. 請複製以下應用程式碼，然後貼到編輯器中：

      -- CREATE OR REPLACE STREAM for cleaned up referrer
      CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
          "ingest_time" TIMESTAMP,
          "referrer" VARCHAR(32));
          
      CREATE OR REPLACE PUMP "myPUMP" AS 
         INSERT INTO "DESTINATION_SQL_STREAM"
            SELECT STREAM 
               "APPROXIMATE_ARRIVAL_TIME", 
               REGEX_REPLACE("REFERRER", 'http://', 'https://', 1, 0)
            FROM "SOURCE_SQL_STREAM_001";

   2. 選擇儲存並執行 SQL。在即時分析標籤上，您可以查看應用程式建立的所有應用程式內串流，並驗證資料。