Kinesis AnalyticsのSQLクエリ

Kinesis AnalyticsをSQLクエリについて、AWS側がテンプレートを用意してくれてるのですが、微妙に間違っていたりするのでまとめました。また、ハマりどころも多かったのでまとめています。

#Kinesis Analyticsを使用する上でのポイント
・Stream名とデータ名の名前は大文字小文字で区別される
 　 → Stream名もしくはデータ名を「"」で括ると小文字になり、括らないと大文字と認識される
　　→ （個人的には）Create Stream文内のStream名とデータ名は大文字がよい
・Kinesis Streams demoは良く停止する
　　→ 一旦削除して、新たに作成したほうがよい
・Reference Dataを追加すると、[Could not find Stream]というエラーが発生する
・SQLコードを[Save and Running]させると実行まで時間が掛かるときがある
　　- Random_CUT関数は30分ほど時間を要する場合がある
・テンプレートが微妙に間違っているときがある
    
#所感
・構築は楽
・KPUはオートスケーリングだが、キャップ機能はないのでお金に注意
・結構、バグがある。コードのupdatingからRUNNINGまで時間が掛かるときがある
・ドキュメントが少ない
・テンプレートが間違っているので注意。高度な分析の関数はまじでわからん・・・



紹介するクエリは下記の通りです。

#基本的なクエリ
・CREATE PUMP & STREAM
・SELECT
・WHERE
・MULTI STREAMS
・GROUP BY
・WINDOW
　1.TUMBLING WINDOW
　2.SLIDING WINDOW
・INNER JOIN


#高度な分析
・RANDOM_CUT_FOREST
・DISTINCT
・GROUP RANK ※できていない
・TOP-K　※できていない


#基本的な構文
## CREATE PUMP & STREAM, SELECT
・[OR REPLACE]はSTREAM名が既に存在すれば置き換えるという意味
・AnalyticsはPUMPにデータを投入後に、STREAMにデータを格納する
・関数名を列名に使用できない（avg、min,maxなど）

```
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM (
   TICKER_STNBOL VARCHAR(4),
   SECTOR        VARCHAR(16),
   PRICE         REAL,
   CHANGE        DECIMAL);

CREATE OR REPLACE PUMP STREAM_PUMP AS INSERT INTO DESTINATION_SQL_STREAM
   SELECT STREAM ticker_symbol, sector, change, price
   FROM "SOURCE_SQL_STREAM_001";
```

<img width="1186" alt="SELECT.png" src="https://qiita-image-store.s3.amazonaws.com/0/132083/4a4cb891-028f-37e1-7ed1-474e379a2e4d.png">


## WHERE
・基本的に標準SQLと同じ

```
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM (
   TICKER_SYMBOL VARCHAR(4),
   SECTOR        VARCHAR(16),
   PRICE         REAL,
   CHANGE        DECIMAL);

CREATE OR REPLACE PUMP STREAM_PUMP AS INSERT INTO DESTINATION_SQL_STREAM
   SELECT STREAM ticker_symbol, sector, change, price
   FROM "SOURCE_SQL_STREAM_001"
   WHERE SECTOR LIKE '%ALTH%' 
;
```
<img width="1181" alt="WHERE.png" src="https://qiita-image-store.s3.amazonaws.com/0/132083/1747d855-5554-9080-ccd8-18e17838f502.png">


##MULTI STREAMS
・PUMP名は一緒にしてはならない
・MULTI STREAMSを定義した場合、AWSコンソールのreal-time analyticsタブ内の「In-application streams」に複数STREAMが表れる

```
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM (
   TICKER_SYMBOL VARCHAR(4),
   SECTOR        VARCHAR(16),
   PRICE         REAL,
   CHANGE        DECIMAL);

CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM2 (
   TICKER_SYMBOL VARCHAR(4),
   SECTOR        VARCHAR(16),
   PRICE         REAL,
   CHANGE        DECIMAL);

CREATE OR REPLACE PUMP STREAM_PUMP AS INSERT INTO DESTINATION_SQL_STREAM
   SELECT STREAM ticker_symbol, sector, change, price
   FROM SOURCE_SQL_STREAM_001
   WHERE SECTOR LIKE '%ALTH%' 
;


CREATE OR REPLACE PUMP STREAM_PUMP2 AS INSERT INTO DESTINATION_SQL_STREAM2
   SELECT STREAM ticker_symbol, sector, change, price
   FROM SOURCE_SQL_STREAM_001
   WHERE PRICE > 0.0
;
```

<img width="1178" alt="MULTISTREAM.png" src="https://qiita-image-store.s3.amazonaws.com/0/132083/83260ffe-5276-edde-925c-a72390c88f41.png">



##GROUP BY（TUMBLING WINDOW)
・構文は通常SQLと同様だが、TIME WINDOWを設定する必要がある（TUMBLING WINDOWと言う）
・ROWTIMEはKinesis Analyticsに予め用意されている時刻で収集した時刻が格納されている

```
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM (
  TICKER_SYMBOL VARCHAR(4), 
  TICKER_SYMBOL_COUNT INTEGER,
  TICKER_SYMBOL_AVG REAL
);

CREATE OR REPLACE  PUMP STREAM_PUMP AS INSERT INTO DESTINATION_SQL_STREAM
SELECT STREAM 
   ticker_symbol, 
   COUNT(*) AS ticker_symbol_count,
   AVG(price) AS ticker_symbol_avg
FROM "SOURCE_SQL_STREAM_001"
GROUP BY ticker_symbol, 
         FLOOR(("SOURCE_SQL_STREAM_001".ROWTIME - TIMESTAMP '1970-01-01 00:00:00') SECOND / 10 TO SECOND);
```

<img width="1190" alt="GROUPBY.png" src="https://qiita-image-store.s3.amazonaws.com/0/132083/4bce2b36-1439-18b6-a9c5-65e571e0fff6.png">


##SLIDING WINDOW
・SLIDING WINDOWを使用する場合、GROUP BY句は使用せず、集約関数(avg,min,count等)の直後にWINDOWを定義する
・WINDOW幅の設定はTIMEとROWの２種類が存在する（微妙に書き方が違うので注意）

```
CREATE STREAM DESTINATION_SQL_STREAM (
    TICKER_SYMBOL VARCHAR(4), 
    TICKER_SYMBOL_AVG_T REAL,
    TICKER_SYMBOL_COUNT_T INTEGER,
    TICKER_SYMBOL_AVG_R REAL,
    TICKER_SYMBOL_COUNT_R INTEGER);
    
CREATE PUMP STREAM_PUMP as insert into DESTINATION_SQL_STREAM
SELECT STREAM ticker_symbol,
              avg(price) over TSW as ticker_symbol_avg1,
              count(*)  over TSW as cnt1,
              avg(price) over RSW as ticker_symbol_avg2,
              count(*)  over RSW as cnt2
FROM SOURCE_SQL_STREAM_001
WINDOW TSW as (partition by ticker_symbol Range interval '10' second preceding),
RSW as (partition by ticker_symbol  ROWS 2 preceding);
```

<img width="1187" alt="GROUP BY SLID.png" src="https://qiita-image-store.s3.amazonaws.com/0/132083/761b3ae2-be60-b465-7f9c-f43c2dcda78c.png">


##INNER JOIN
・異なるSTREAM同士のJoin処理は可能
・Reference Dataを追加する動作しない(2017/3月時点）

```
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM3 (
  TICKER_SYMBOL VARCHAR(4),
  TICKER_SYMBOL_COUNT INTEGER, 
  TICKER_SYMBOL_AVG REAL);

CREATE OR REPLACE PUMP STREAM_PUMP3 AS INSERT INTO DESTINATION_SQL_STREAM3
SELECT 
   STREAM A.ticker_symbol,  
          A.ticker_symbol_count,
          B.ticker_symbol_avg
FROM DESTINATION_SQL_STREAM as A
INNER JOIN DESTINATION_SQL_STREAM2 as B
ON A.ticker_symbol = B.ticker_symbol
```

<img width="1198" alt="INNER JOIN.png" src="https://qiita-image-store.s3.amazonaws.com/0/132083/f7e1c758-1140-8aad-604d-deb578067a6f.png">


####INNER JOINで使用しているDESTINATION_SQL_STREAMとDESTINATION_SQL_STREAM2の作成

```
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM (
  TICKER_SYMBOL VARCHAR(4), 
  TICKER_SYMBOL_COUNT INTEGER
);

CREATE OR REPLACE  PUMP STREAM_PUMP1 AS INSERT INTO DESTINATION_SQL_STREAM
SELECT 
  STREAM ticker_symbol, 
         count(*) AS ticker_symbol_count
FROM "SOURCE_SQL_STREAM_001"
GROUP BY ticker_symbol, 
         FLOOR(("SOURCE_SQL_STREAM_001".ROWTIME - TIMESTAMP '1970-01-01 00:00:00') SECOND / 10 TO SECOND);

CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM2 (
  TICKER_SYMBOL VARCHAR(4), 
  TICKER_SYMBOL_AVG REAL
);

CREATE OR REPLACE  PUMP STREAM_PUMP2 AS INSERT INTO DESTINATION_SQL_STREAM2
SELECT 
  STREAM ticker_symbol, 
         avg(price) AS ticker_symbol_avg
FROM "SOURCE_SQL_STREAM_001"
GROUP BY ticker_symbol, 
         FLOOR(("SOURCE_SQL_STREAM_001".ROWTIME - TIMESTAMP '1970-01-01 00:00:00') SECOND / 10 TO SECOND);
```


####Reference Dataの追加
・AWS CLIでのみ追加が可能
・current-application-version-idはdescribe-applicationで確認する

```
aws kinesisanalytics add-application-reference-data-source \
--region us-east-1   \
--application-name [your-kinesis-analytics-name] \
--current-application-version-id [?] \
--reference-data-source '{"TableName":"companyname","S3ReferenceDataSource":{"BucketARN":"arn:aws:s3:::[バケット名]","FileKey":"[ファイル名]","ReferenceRoleARN":"arn:aws:iam::[アカウントID]:role/service-role/[ロール名]"},
"ReferenceSchema":{ "RecordFormat":{"RecordFormatType":[CSV or json], "MappingParameters":{"CSVMappingParameters":{"RecordRowDelimiter":"\n","RecordColumnDelimiter":","} }},
"RecordEncoding":"UTF-8","RecordColumns":[{"Name":"type","SqlType":"varchar(5)"},{ "Name":"company","SqlType":"varchar(10)"}]}}'
```
####アプリケーションの確認
・aws configureで事前に設定しておくこと

```
aws kinesisanalytics describe-application --application-name [your-kinesis-analytics-name]
```
####Reference Dataの削除
・version-idとreference-idはdescribe-applicationで確認する

```
aws kinesisanalytics delete-application-reference-data-source --application-name [your-kinesis-analytics-name] --current-application-version-id [?] --reference-id [?]
```

#高度な分析
・高度な分析ではCursol関数を使用。（標準SQLと結構違うので最初はだいぶ戸惑う）
・書き方は「SELECT STREAM * FROM(TABLE(xxxFUNCTION(COURSOR([in-application-stream],option)));」

##RANDOM_CUT_FOREST(異常検知）
・記載する情報は「in-application-stream」、「numberOfTrees(default:100)」、 「subSampleSize(default：256)」、「timeDecay(default：100,000)」、 「shingleSize (default：1)」の５つ
・詳細はマニュアルを・・・（正直あまり理解していない）

```
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM (
   TICKER_SYMBOL    VARCHAR(4),
   SECTOR VARCHAR(10),
   PRICE   DOUBLE,
   CHANGE DOUBLE,
   ANOMALY_SCORE  DOUBLE);

CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM2 (
   TICKER_SYMBOL    VARCHAR(4),
   SECTOR VARCHAR(10),
   PRICE   DOUBLE,
   CHANGE DOUBLE,
   ANOMALY_SCORE  DOUBLE);

CREATE OR REPLACE PUMP STREAM_PUMP AS INSERT INTO DESTINATION_SQL_STREAM
SELECT STREAM  ticker_symbol,sector,price,change,ANOMALY_SCORE FROM
      TABLE(RANDOM_CUT_FOREST(
        CURSOR(SELECT STREAM * FROM "SOURCE_SQL_STREAM_001")
      )
    ) ;

CREATE OR REPLACE PUMP STREAM_PUMP2 AS INSERT INTO DESTINATION_SQL_STREAM2
SELECT 
   STREAM * FROM DESTINATION_SQL_STREAM
ORDER BY FLOOR(DESTINATION_SQL_STREAM.ROWTIME TO SECOND), 
         ANOMALY_SCORE DESC;
```
<img width="1170" alt="RANDOMCUT.png" src="https://qiita-image-store.s3.amazonaws.com/0/132083/18af3e97-4747-68c3-abd0-b6ae1e27c50e.png">


##DISTINCT COUNT
・COUNT_DISTINCT_ITEMS_TUMBLING関数に記載する情報は「in-application-stream」、「列名」、「TUMBLING WINDOW時間」の3つ。列名は１つのみで複数列の指定はできない
・出力される情報はDistinct Countした情報のみ 

```
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM (NUMBER_OF_DISTINCT_ITEMS BIGINT);
CREATE OR REPLACE PUMP STREAM_PUMP AS INSERT INTO DESTINATION_SQL_STREAM
SELECT STREAM NUMBER_OF_DISTINCT_ITEMS FROM TABLE(COUNT_DISTINCT_ITEMS_TUMBLING(
  CURSOR(SELECT STREAM * FROM "SOURCE_SQL_STREAM_001"),
  'TICKER_SYMBOL', -- name of column in single quotes
  10 -- tumbling window size in seconds
  )
);
```

<img width="1182" alt="Distinct.png" src="https://qiita-image-store.s3.amazonaws.com/0/132083/da00d9d5-7cf4-9941-fba4-b0faa77304b3.png">


##GROUP RANK
・すみません、いまのところエラーが出て不明。（ドキュメント読んだのだが・・・）
・エラー内容は「SQL error message: From line 2, column 55 to line 9, column 3: No match found for function signature GROUP_RANK(<CURSOR>, <CHARACTER>, <CHARACTER>, <CHARACTER>, <CHARACTER>, <CHARACTER>, <CHARACTER>)」
・使用したクエリは下記。
・情報があれば共有してほしいです

```
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM (
    TICKER_SYMBOL VARCHAR(4),
    PRICE real,
    RANK_NUM integer
);

CREATE OR REPLACE PUMP STREAM_PUMP AS INSERT INTO DESTINATION_SQL_STREAM
SELECT STREAM ticker_symbol,price,rank_num FROM TABLE(GROUP_RANK(
  CURSOR(SELECT STREAM ticker_symbol,price FROM "SOURCE_SQL_STREAM_001"),
  'PRICE', -- rankByColumnName
  'rank_num', -- rankOutColumnName
  'desc', -- sortOrder
  'asc', --outputOrder
  10, --maxIdle
  5 --outputMax
  )
);
```

##TOP-K
・ドキュメントを読んで、下記の通りで記載したが、SELECT STREAM後のticker_symbolがないというエラーが発生
・ただ、テンプレートにあるクエリは回っていて、TOP-K関数からでてくる値はITEM,ITEM_COUNTという謎の2列のようです。（詳細についてドキュメントがされず、色々と試したけど何も出力されてないので）
・情報があれば共有してほしいです

```
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM(
    TICKER_SYMBOL VARCHAR(4), 
    PRICE REAL)
;

CREATE OR REPLACE PUMP STREAM_PUMP AS INSERT INTO DESTINATION_SQL_STREAM
SELECT STREAM ticker_symbol, price FROM TABLE(TOP_K_ITEMS_TUMBLING(
  CURSOR(SELECT STREAM * FROM "SOURCE_SQL_STREAM_001"),
  'ticker_symbol', -- name of column in single quotes
  10, -- number of top items
  10 -- tumbling window size in seconds
  )
);
```

いまのところ、やってみたクエリを整理して紹介しました。