[Oracle Cloud] Data Integration でファイル分割してみた

# はじめに

Oracle Cloud Integration では、ETL 機能を提供する Data Integration サービスがあります。Data Integration は、様々なデータアセットからデータを取り込み、データのクレンジング・変換・再形成・変換などの ETL 処理が出来るフルマネージドサービスです。サーバー管理が不要で、GUI で視覚的にわかりやすく ETL 処理を表現できます。

ファイルの分割方法を簡単に紹介します。ファイル内の特定の列の値によって、出力するファイルを分割したい用途を想像しています。具体例は次の通りです。



データ変換前です。Object Storage に `user-split.csv` というファイルがあります。

```
name,userrank,company
sugiyama,bronze,a
tanaka,silver,b
kimura,gold,c
satou,bronze,a
koga,silver,b
komikado,gold,c
suzuki,bronze,a
eto,silver,b
kato,platinum,c
```



company の列は、a, b, c と3種類の会社があります。この3種類の会社ごとにファイルを分割して出力する方法を紹介します。



# 留意点

Data Integration 側で、受け取ったデータの company を解析して、ファイルを動的に分割は出来ません。Data Integration のタスクを実行側で「companyには、このデータが存在しているはずだから、この名前で分割をしてね」とパラメータを与えることで、想定した動きに出来ます。



# データ分割する Data Flow を作成

まず、Input の csv ファイルを定義します。

![1600731339785.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/8e3903f4-e163-dfd3-d4f9-3c7f42bcc63c.png)



Company Name で Filter を設定する部分です。Parameter 化をするので、わかりやすい説明を入れています。

![1600731407922.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/34b7027b-d48a-5dac-2530-8ecebb98b786.png)



Data Tab を確認し、エラーが発生しないことを確認します。Filter の条件を `please input company name` としているため、No items found となるのは正常です。

![1600731494622.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/78da466a-553f-b4ff-a1b7-b1f3f5f13c23.png)



Filter に、Parameter を Assign します

![1600731566013.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/8ad75bc3-7f7c-b2fd-d155-f9ddcaa5067c.png)



Add New Parameter を押します

![1600731597681.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/1bf4c170-cc2b-2c95-91de-63245b712c53.png)



Identifier にわかりやすい名前を入れて、Add をします

![1600731636741.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/3ec779e3-5392-cdaa-2161-f0123d9e5f22.png)



Assign します

![1600731682787.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/b7f50cfc-6fd5-cb5e-7fd3-3a64e76adc6f.png)



表示が変わります

![1600731707375.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/4b4899ac-01eb-8972-ae57-fca2847bfe73.png)



Target の Object Storage の設定を入れます

![1600731821378.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/d61b9a9d-9f79-f6a8-fa43-b814f7a310ad.png)



Output Bucket と、そこに作成する Directory Name を指定

![1600731922702.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/5fb20094-c06c-1991-18de-ce1b85c69d86.png)



最後に、Validate して、Save and Close をします

![1600731967621.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/10ceda29-7fb3-5b55-f413-18aab6ea7cce.png)



# 動作確認

作成した Data Flow を、Integration Task として作成して、Application に Publish した上で、動作確認してみます

![1600732181922.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/1593875a-c34e-424f-98b2-5cddebb1d284.png)



Parameter で指定した名前が、Edit 出来るようになっています。Edit を押します。

![1600732217708.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/01bea319-ad4d-488b-7994-2f277b50bd4d.png)



Condition をそのまま編集できる画面になっています

![1600732254220.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/4f0eb765-36fa-c7cf-2cd1-fafb8c01ecf8.png)



Company Name が `a` をフィルターしてみます

![1600732294413.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/0bd70b86-fd07-d462-e51d-a46e5fb76809.png)



Runします

![1600732335753.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/f97a19e8-c3c5-7751-7c9f-5927c2c61635.png)



Task Runs の一覧に出力されています。一定時間後に、Success となります。

![1600732399667.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/41ee7896-fd32-f4b0-2830-613d5c67bb4e.png)



output bucket に出力されています

![1600732469184.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/6e667270-3feb-612a-4446-44e16b5e9fa3.png)



ファイルの中身を見てみると、こんな感じになっています。正しく `a` でフィルターされていますね。

```
name,userrank,company
sugiyama,bronze,a
satou,bronze,a
suzuki,bronze,a
```



同じような方法で、パラメータに、b や c を入力すると、意図したファイル分割が出来ます。



# 想定外のデータ出力

今回の方法は、Task の実行側で Filter の条件を指定する方法です。`a` や `b`, `c` Company ということが分かっているので、指定しています。ただ、`e` といった想定外のデータが入ったときには、それを検知したいと思うはずです。データ加工のタスクを実行する際に、次のようにパラメータを与えるのが良いと思います



- 欲しいデータ。次の名前でフィルターする
  - `a`
  - `b`
  - `c`
- 想定外のデータ。検知はしたい
  - `a`, `b` , `c` 以外のデータをフィルター




想定外のデータは、想定外専用の Object Storage を用意して、そこに出力すると良いでしょう。例を挙げます。次のデータを与えたときに、一番下の `d` company に所属している erroruser を検知したいと思うはずです。

```
name,userrank,company
sugiyama,bronze,a
tanaka,silver,b
kimura,gold,c
satou,bronze,a
koga,silver,b
komikado,gold,c
suzuki,bronze,a
eto,silver,b
erroruser,hoge,d
```



これを実現するための方法を確認してみましょう。Task Run する時のパラメータで、次の Condition を指定すると、`a`,`b`,`c` のどれでもないデータを抽出できます

```
NOT FILTER_1.SOURCE_1.company = 'a'
AND
NOT FILTER_1.SOURCE_1.company = 'b'
AND
NOT FILTER_1.SOURCE_1.company = 'c'
```

![1600734741270.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/0e3da9ef-42c9-dff2-f61c-293cf4255377.png)



上記で実行したときの実行例です。

![1600734824560.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/12cf1f81-6a18-b614-6253-3f926f5c8350.png)



csv ファイルの中身です

```
name,userrank,company
erroruser,hoge,d
```



実際の運用では、エラーデータ格納用の Object Storage Bucket を用意して、ファイルが格納されたことを検知して Functions を起動させ、ファイルの中身にデータがあれば、Notification などで Email や Slack 通知をすると良いでしょう。



# 付録 : 指定した Company Name がなかったとき

a, b, c といった Company Name を指定してフィルターをします。この時、指定した名前のデータが無かった時はどのような挙動になるでしょうか？ 結論は、Task が正常に終了し、中身が空のデータが出来上がります。見てみましょう。

`e` でフィルターしてみます。`e` のデータは無いので、中身が空のデータが出来上がる予定です。

![1600735280767.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/ca3280e9-0469-a669-f107-971e34be6302.png)



Task が成功します。

![1600735379057.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/c4e3c0d1-3fa1-1b33-86cf-baf970ae2d72.png)



中身は空ですが、Directory と空のファイルが出来上がります

![1600735405802.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/6a01e1bc-3db1-12b7-f666-7e9a01ccfe3a.png)