StreamSetsのDataCollectorを使用してGUIでデータフローを定義する

Cloudera Managerを使ってセットアップしたData CollectorのGUIからデータフローを作成します。
<img width="1386" alt="t26.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/335d0e70-889b-42b9-8ed2-1a9b3c902612.png">


StreamSetsのドキュメントにあるチュートリアルに従って作ってみます。今回はBasic Tutorialのパイプラインを作ってみましょう。
https://streamsets.com/documentation/datacollector/latest/help/#Tutorial/Tutorial-title.html

# テストデータのダウンロード

チュートリアルに従ってディレクトリを作成し、サンプルデータをダウンロードします。

```
[cloudera@quickstart ~]$ sudo mkdir /streamsets
[cloudera@quickstart ~]$ mkdir -p /streamsets/tutorial/origin
[cloudera@quickstart ~]$ mkdir -p /streamsets/tutorial/destination
[cloudera@quickstart ~]$ mkdir -p /streamsets/tutorial/error
[cloudera@quickstart ~]$ sudo chmod -R 777 /streamsets
[cloudera@quickstart ~]$ cd /streamsets/tutorial/origin/
[cloudera@quickstart origin]$ wget https://www.streamsets.com/documentation/datacollector/sample_data/tutorial/nyc_taxi_data.csv
--2016-07-20 01:56:40--  https://www.streamsets.com/documentation/datacollector/sample_data/tutorial/nyc_taxi_data.csv
Resolving www.streamsets.com... 162.242.235.205
Connecting to www.streamsets.com|162.242.235.205|:443... connected.
HTTP request sent, awaiting response... 301 Moved Permanently
Location: https://streamsets.com/documentation/datacollector/sample_data/tutorial/nyc_taxi_data.csv [following]
--2016-07-20 01:56:41--  https://streamsets.com/documentation/datacollector/sample_data/tutorial/nyc_taxi_data.csv
Resolving streamsets.com... 162.242.235.205
Connecting to streamsets.com|162.242.235.205|:443... connected.
HTTP request sent, awaiting response... 200 OK
Length: 1054132 (1.0M) [text/csv]
Saving to: “nyc_taxi_data.csv”

100%[======================================>] 1,054,132    828K/s   in 1.2s    

2016-07-20 01:56:43 (828 KB/s) - “nyc_taxi_data.csv” saved [1054132/1054132]

[cloudera@quickstart origin]$ 
```

# 基本チュートリアル

チュートリアルに従って進めていきます。

## パイプラインの作成とパイプラインのプロパティの定義

まずはGUIからパイプラインを作成してみます。Data Collector の Web UIから「Create New Pipeline」をクリックします。

<img width="1386" alt="ss_top2.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/957f19fd-1c2b-dbfc-e28e-281ec7478c08.png">

パイプラインの名前は Tutorial1 としてみました。

<img width="536" alt="t1.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/f7c06700-48cf-b4d4-8814-d02b5cc1562c.png">

「Save」を押して保存すると、空のキャンバスが表示されます。

<img width="1384" alt="t2.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/c3c11b44-bda5-0f26-a36e-89cdc8b103d6.png">

キャンバスの説明はドキュメントに書かれています。
https://streamsets.com/documentation/datacollector/latest/help/#Pipeline_Configuration/DataCollectorWindow-Config.html

## Directory Origin を設定する

パイプラインの「入り口」に相当するのがOriginです。今回はダウンロードしたサンプルのCSVファイルがあるディレクトリをOriginにします。ドロップダウンメニューの「Select Origin」 -> 「Directory」を選択するか、右側のStage LibraryからDirectory Originを選びます。

<img width="477" alt="t3.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/7939d4d5-73a1-2421-d41d-905b0d450436.png">

<img width="169" alt="t4.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/58cd19c0-27a3-18c2-0c36-2aa81cde6de3.png">

キャンバスに Directory Originが配置されました。アイコンをクリックし、下の画面からFilesタブをクリックし、次のパラメータを指定します。

- Data Format: Delimited
- Files Directory: /streamsets/tutorial/origin
- Read Order: Last Modified Timestamp
- File Name Pattern: *taxi*.csv
- First File to Process: 空のまま

Data FormatにDelimitedを指定するとDelimitedタブで詳細設定ができます。今回は以下のパラメータを指定します。

- File Type: Default CSV
- Header Line: With Header Line (CSVの先頭行にヘッダ行があるので）
- Root Field Type: List-Map

## データのプレビュー

Originで登録したデータをプレビューしてみましょう。データが正しいかどうかをGUIから「見る」ことができるのは便利ですね。

チュートリアルによれば、この状態でPreviewアイコンをクリックできるようですが、手元の環境ではうまくいきませんでした。左側の❗️をクリックしてエラーを解決します。今回はCREATION_009というエラーが表示されています。

<img width="1233" alt="t7.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/20dc9a22-2afe-4c9d-d753-2445161b64ae.png">

エラーデータの扱いを指定しなければならないようなので、まずは無視することにしました。

- Error Record: Discard (Libary Basic) 

今度はPreviewアイコンがクリックできるようになっているので、クリックしてプレビューしましょう。

<img width="520" alt="t9.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/0f0257c5-e838-a6a4-3ee1-3b289727ddd1.png">

<img width="1223" alt="t10.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/8db1994e-d337-84b5-671d-4e4cf20aa12c.png">

出力結果は以下のようになりました。CSVファイルの各列をリスト形式で表示しています。

<img width="666" alt="t11.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/e51114ac-64b8-6f9f-7e3c-62401d9791a2.png">

テーブル形式でも表示できます！これ、CSVファイルをエディタなどで見る必要ないですし、ヘッダ行も表示してくれていますし、便利ですねぇ....

<img width="1385" alt="t12.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/ff81369d-1077-493f-7fe2-d1e927c07f67.png">

今回のチュートリアルでは、支払いタイプ(/payment_type)とクレジットカード情報(/credit_card)列を利用するようです。終わったら Close Preview アイコンをクリックします。

<img width="69" alt="t13.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/40c7a251-7422-1a2f-826c-44eeaf8cf041.png">

## Stream Selector でデータの経路を作る

「Select Processor to Connect」->「Stream Selector」を選択します。

<img width="506" alt="t14.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/1e853e46-7af6-69b9-8786-98b8bf54f161.png">

Stream Selector の General タブで、Required Fields テキストボックスを指定します。

<img width="1218" alt="t15.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/2f5cfc1d-9a30-1f19-50b9-abd5339aac96.png">

今回は支払いの種類がないレコードをdiscardしたいので、以下の設定を行います。

- Required Fields: /payment_type

Conditionタブで条件を指定します。

条件により経路が変わります。今回は条件を満たしたデータ(支払いの種類がクレジットカード）が出力ストリーム（１）に流れていくことになります。満たさなかったデータは(2)に流れます。条件の指定の構文を覚える必要はありますが、自動補完が良くできているので、割とサクサク入力できる印象です。

- ${record:value('/payment_type')=='CRD'}

<img width="1224" alt="t16.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/02df90ce-e1e0-8f5a-6907-7762f748cbcc.png">

２系統の出力があることがわかりますね。

## Jythonを使用してクレジットカードの種類を取得する

ストリーム中のクレジットカード情報のデータを扱うため、Jythonのスクリプトを使用します。「Select processor to connect」からJython Evaluator processorを追加します。

<img width="685" alt="t17.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/2ff11de6-f8d3-8640-dd36-5b49808e3d2d.png">

続いでJythonタブのスクリプトテキストボックスに、チュートリアルのJythonスクリプトを貼り付けます。

<img width="1203" alt="t18.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/b371a848-fb10-3f1e-7462-fc11106a168b.png">

## クレジットカード番号をマスクする

セキュリティ上、クレジットカード番号をそのまま利用するのは避けたいものです。Data Collectorにはマスクの機能もあるのでデータフローでカード番号をマスクします。先ほどと同様の手順で、キャンバスにField Maskerを追加します。以下のプロパティを指定し、正規表現を使ってマスクします。

- Mask Type: Regular Expression
- Regular Expression: (.*)([0-9]{4})
- Groups to Show: 2

<img width="1226" alt="t19.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/5be5db68-ed9c-d324-a5e0-cafcd2a15d5e.png">


## Destination に書き込む

パイプラインの結果の出力先を指定します。チュートリアルではローカルファイルシステムになっていますが、せっかくのCloudera Manager管理しているHDFSがあるわけですから今回はHDFSを指定しようかと思ったのですが、Kerberos化していたの後日試すことにしました。「Select Destination to Connect」から「Hadoop FS - CDH5.7.1」を選択します。

<img width="780" alt="t20.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/fce91f25-10bf-e997-3814-261011f196e9.png">

Hadoop FS Propertyは、暫定的に書き込み先を file:///にします。

- Hadoop FS URI: file:///
- HDFS user: cloudera

<img width="1220" alt="t21.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/5a92a635-5b6e-b245-512a-63818362adca.png">

Output Filesタブで以下のプロパティを指定します。

- Data Format: Delimited
- File Prefix: _out
- Directory Template: /streamsets/tutorial/destination
- Data Timezone: JST

Delimitedタブで以下のプロパティを設定します。

- Header Line: With Header Line

## Expression Evaluatorで対応するフィールドを追加する

先ほどのJython Evaluatorではクレジットカード情報のデータを処理し、クレジットカードの種類を追加しています。ここでデータをプレビューしてみましょう。先ほどと同じように「Preview」をクリックし、Jython Evaluatorをクリックします。
入力データと出力データが表示されます。期待通り（？）、出力にはcredit_card_typeのデータが含まれていますね！

<img width="1382" alt="t22.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/50199e2e-dd7d-82ad-d618-05e68ce524e2.png">

クレジットカード支払いでなかった場合も考慮する必要があるので、Stream Selectorのdefaultの出力に「Expression Evaluator」を追加します。

<img width="1222" alt="t23.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/97bbdd2e-0a93-d728-8612-66c24fc5c0fe.png">

Expressionタブのプロパティを以下のように指定します。

- Output Field: n/a
- Header attributeは削除

終わったら、Expression EvaluatorとHadoop FSをつなぎます。

<img width="1221" alt="t24.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/aa3ed875-76b3-4525-15ca-52774e910936.png">

## データルールとアラートの作成

さて、いよいよ最後です。Stream SelectorからJython Evaluatorの間の線をクリックしてルールを追加します。今回はクレジットカードデータがない（エラー）の場合のルールを追加をしています。

Data Ruleのプロパティ
- Label: Missing Card Numbers
- Condition: ${recod:value("/credit_card")==""}
- Sampling Percentage: 35
- Alert text: At least 10 missing credit card numbers!
- Threshold Value: 10


#パイプラインを実行する

いよいよ実行です。▶︎のアイコンをクリックして実行します。

<img width="1386" alt="t26.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/a490eeca-dedd-9cb8-6cbd-dafc78573408.png">

クレジットカードのエラーは40件のようです。

<img width="1367" alt="t27.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/11571611-23b7-6749-1e73-af7cb9cba732.png">

出力されたファイルも見てみましょう。クレジットカード番号はマスクされ、最後の列にカードの種類が出力されています。

```
[cloudera@quickstart origin]$ head /streamsets/tutorial/destination/_tmp__out
F6F7D02179BE915B23EF2DB57836442D,088879B44B80CC9ED43724776C539370,VTS,CRD,12,0.5,0.5,1.75,0,14.75,1,2013-01-13 04:36asdf,2013-01-13 04:46asdf,5,600,3.12,-73.996933,40.720055,-73.993546,40.693043,xxxxxxxxxxxx2922,Visa
BE386D8524FCD16B3727DCF0A32D9B25,4EB96EC9F3A42794DEE233EC8A2616CE,VTS,CRD,12,0.5,0.5,3.12,0,16.12,1,2013-01-13 04:37:00,2013-01-13 04:48:00,2,660,3.39,-74.000313,40.730068,-73.987373,40.768406,xxxxxxxxxxxx0902,MasterCard
E9FF471F36A91031FE5B6D6228674089,72E0B04464AD6513F6A613AABB04E701,VTS,CRD,5.5,0.5,0.5,1.2,0,7.7,1,2013-01-13 04:41:00,2013-01-13 04:45:00,1,240,1.16,-73.997292,40.720982,-74.000443,40.732376,xxxxxxxxxxxx9608,Visa
A5D125F5550BE7822FC6EE156E37733A,08DB3F9FCF01530D6F7E70EB88C3AE5B,VTS,CRD,11,0.5,0.5,2,0,14,1,2013-01-13 04:37:00,2013-01-13 04:47:00,5,600,2.91,-73.966843,40.756741,-73.987885,40.722713,xxxxxxxxxxxx7252,Visa
EE1513D432B07F7E0B5E2ED1EF629086,F31D261881520931062C011366E56A04,VTS,CRD,18,0.5,0.5,3.7,0,22.7,1,2013-01-13 04:30:00,2013-01-13 04:44:00,6,840,5.21,-74.005455,40.740772,-73.967354,40.798096,xxxxxxxxxxx7913,AMEX
```


# まとめ

正直に言ってかなり使い易いと思います。HDFS、HBase、Kafka等、様々なエコシステムへのパイプラインを作成できるので、一度試してみてはいかがでしょうか？


## おまけ
Cloudera ManagerでData Collectorのモニタリング
<img width="1379" alt="t28.png" src="https://qiita-image-store.s3.amazonaws.com/0/13776/9f741d72-c221-2c7b-2a4d-26761d267215.png">