範例：轉換多個資料類型

首先，將串流來源對應到應用程式內輸入串流，類似於所有其他 Kinesis Data Analytics 應用程式。

然後，在應用程式碼中撰寫 SQL 陳述式，從應用程式內輸入串流擷取特定類型的資料列。接著將它們插入個別的應用程式內串流。(您可以在應用程式碼中建立其他應用程式內串流。)

將訂單和交易列過濾到個別的應用程式內串流：

1. 篩選 SOURCE_SQL_STREAM_001 中的訂單記錄，並將訂單儲存在 Order_Stream 中。

   --Create Order_Stream.
   CREATE OR REPLACE STREAM "Order_Stream" 
              ( 
               order_id     integer, 
               order_type   varchar(10),
               ticker       varchar(4),
               order_price  DOUBLE, 
               record_type  varchar(10)
               );
   
   CREATE OR REPLACE PUMP "Order_Pump" AS 
      INSERT INTO "Order_Stream"
         SELECT STREAM oid, otype,oticker, oprice, recordtype 
         FROM   "SOURCE_SQL_STREAM_001"
         WHERE  recordtype = 'Order';

2. 篩選 SOURCE_SQL_STREAM_001 中的交易記錄，並將訂單儲存在 Trade_Stream 中。

   --Create Trade_Stream.      
   CREATE OR REPLACE STREAM "Trade_Stream" 
              (trade_id     integer, 
               order_id     integer, 
               trade_price  DOUBLE, 
               ticker       varchar(4),
               record_type  varchar(10)
               );
   
   CREATE OR REPLACE PUMP "Trade_Pump" AS 
      INSERT INTO "Trade_Stream"
         SELECT STREAM tid, toid, tprice, tticker, recordtype
         FROM   "SOURCE_SQL_STREAM_001"
         WHERE  recordtype = 'Trade';

現在，您可以對這些串流執行其他分析。在此範例，於一分鐘的翻轉視窗中計算股票代碼的交易數量，並將結果保存到另一個 DESTINATION_SQL_STREAM 串流中。

--do some analytics on the Trade_Stream and Order_Stream. 
-- To see results in console you must write to OPUT_SQL_STREAM.

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
            ticker  varchar(4),
            trade_count   integer
            );

CREATE OR REPLACE PUMP "Output_Pump" AS 
   INSERT INTO "DESTINATION_SQL_STREAM"
      SELECT STREAM ticker, count(*) as trade_count
      FROM   "Trade_Stream"
      GROUP BY ticker,
                FLOOR("Trade_Stream".ROWTIME TO MINUTE);

您會看到下列結果。