watsonx.data: 同じフォーマットの大量のCSVファイルの取り込み

オブジェクトストレージに置いたセンサーデータなど同じフォーマットの大量のCSVファイルを何しから分析したくなったとき、watsonx.data(IBM Cloud SaaS版)でSparkエンジンを使って簡単にSQL Queryの使えるICEBERGテーブルに取り込むことができます。
尚、この投稿の作成時（2023年12月20日）は別サービスのIBM Analytics Engine(Spark)エンジンが必須でしたが、watsonx.data(IBM Cloud SaaS版)内部にSparkエンジンができましたので、別サービスは必須ではになくなりました。

当記事はその方法を説明します。

ここでは[「環境省大気汚染物質広域監視システム　そらまめくん」](https://soramame.env.go.jp/)からダウンロードできる大気汚染測定結果（時間値）のCSVファイル1ヶ月分を取り込んでみます。

> 2024年06月11日に内部のSparkエンジンを使用する方法を書き加えました

# 0. 前提

- IBM watsonx.dataのインスタンスがあること 
- 取り込むCSVデータをおくIBM Cloud Objectストレージ　(S3でもよい）のバケットがIBM watsonx.dataに登録済みであること
- watsonx.dataのカタログがカタログ・タイプ「Apache Iceberg」で1つ以上登録済みであること
- アクセスするuseridにwatsonx.dataに対してIAMで MetastoreAccess roleがあること(2024/6/11 New)

**Sparkにwatsonx.dataのSpark(native spark)を使用する場合**
- Sparkエンジンが追加されていること
  - ない場合は追加する
    - 　参考：[watsonx.data: Sparkエンジンの追加](https://qiita.com/nishikyon/items/0cb71bbc8d89b647ea54)


**SparkにIBM Analytics Engineを使用する場合**
- IBM Analytics Engineのインスタンス があること
  - ない場合は「[1. IBM Analytics Engine Spark インスタンスの作成](https://qiita.com/nishikyon/items/fdaf99066cece2b10ac8)」を参考に作成してください。
<p>

- エンジンに「IBM Analytics Engine(Spark)」　のエンジンが追加されてること
    - 追加されてない場合は[watsonx.data: IBM Analytics Engine(Spark)エンジンを使用したデータ取り込みジョブの作成](https://qiita.com/nishikyon/items/6fd5bdf690632a136b55)の「[2. 前準備2: エンジンに「IBM Analytics Engine(Spark)」　のエンジンを追加](https://qiita.com/nishikyon/items/6fd5bdf690632a136b55#2-%E5%89%8D%E6%BA%96%E5%82%992-%E3%82%A8%E3%83%B3%E3%82%B8%E3%83%B3%E3%81%ABibm-analytics-enginespark%E3%81%AE%E3%82%A8%E3%83%B3%E3%82%B8%E3%83%B3%E3%82%92%E8%BF%BD%E5%8A%A0)」を参考に追加してください


# 1. 大気汚染測定結果（時間値）のCSVファイル1ヶ月分ダウンロード
[「環境省大気汚染物質広域監視システム　そらまめくん」](https://soramame.env.go.jp/)のダウンロードページ
https://soramame.env.go.jp/download にアクセスします。
年月指定「2023年11月」、都道府県「全国」で指定し、zipファイルをダウンロードします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/7b31dfec-bb11-50f0-078a-8cad98b45427.png)

# 2. ダウンロードしたzipファイルを解凍し、フォルダーごとIBM watsonx.dataに登録済みバケットにアップロード
2023年11月の全国のデータをダウンローするとdata.zipファイルに、1708のCSVファイルが入っていました。

文字コードはSJISで改行コードはCR/LFでした。
尚、watsonx.dataで対応しているCSVファイルの文字コードは、
- UTF-8
- UTF-16
- UTF-32

です。幸いヘッダーのみに日本語が入っており、データは半角英数字記号のみで、データの文字コードはSJISでもUTF-8でも変わらないため、このまま使用することにします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/0dc51637-4fa0-49f5-6b42-79b9ddd7d88b.png)

オブジェクトストレージに、フォルダーごとアップロード
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/bea7756d-6b44-70cb-34a8-8929911faeac.png)
フォルダの中身:
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/9ec49590-7e5d-6a89-2690-a00494b80524.png)

# 3. 取り込み先テーブルの作成
Icebergのカタログ上でスキーマを作成後、以下のDDLでテーブルを作成してください。
`lakehouse`は自分のカタログ名、`sensor`は自分のスキーマ名に変更してください。

尚`DATA_DATE`は本来ならばDATE属性にすべきですが、元ファイルのフォーマットが`YYYY/MM/DD`でこのままだとDATE属性では入らないため、一旦varchar属性にしてあります。必要に応じて、DATE属性で別テーブルをCTAS(CREATE TABLE AS SELECT)で作成する等を検討してください。

```sql
CREATE TABLE lakehouse.sensor.jpnair (
   "LOC_CODE" varchar,
   "DATA_DATE" varchar,
   "HOUR" integer,
   "SO2" double,
   "NO" double,
   "NO2" double,
   "NOX" double,
   "CO" double,
   "OX" double,
   "NMHC" double,
   "CH4" double,
   "THC" double,
   "SPM" double,
   "PM25" double,
   "SP" double,
   "WD" double,
   "WS" double,
   "TEMP" double,
   "HUM" double
)
WITH (
   format = 'PARQUET'
)
```

# 4. 取り込みジョブの作成

### 4-1. 「データの取り込み」画面の表示
watsonx.dataの左側のメニューから「データ・マネージャー」のアイコンをクリックし、「データ・マネージャー」を表示します。次に「取り込みジョブの作成」をクリックします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/78145746-1d48-26f5-07fe-75ca8ec2a791.png)


### 4-2. 「エンジン」情報の入力
**エンジン:**
使用するエンジンを選択します。「2. 前準備2」に登録したエンジン名を選びます。

**ジョブ・リソースの構成:**
サイズを選びます

最後に「次へ」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/a2eff6cf-4830-3d2a-b77b-6d789ab49944.png)

### 4-3. 「Select file(s)」情報の入力
「Select remote files」 をクリック
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/87f89d91-1aba-f1b2-7872-c8a6c80e2287.png)

**Bucket**
2でアップロードしたCOSを選択

**Select file type**
ここでは`CSV'を選択


**ソース・ディレクトリー:**
2でアップロードした　**フォルダ**にチェックを入れる。

上記をセットしたら「Next」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/245449bb-4957-1f36-f19b-32df214bb56c.png)


**CSVファイル構成**
- 「ヘッダーあり」　にチェック
- Encoding
  - CSVファイルの文字コードはSJISでしたがデータ部分はUTF-8とコードの変わらない半角英数字記号なので、`UTF-8`にセットします
- Line delimiter
  - `\r\n` にセットします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/5a055a03-fdd6-7edb-e097-a27b49e8cedc.png)

最後に「次へ」をクリックします。


### 4-4. 「ターゲット」情報の入力
取り込み先のテーブルの以下を入力します。「3. 取り込み先テーブルの作成」で作成したテーブルの情報になります。
- カタログ名
- スキーマ名
- テーブル名

最後に「次へ」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/7bc6c4ab-78b9-01ea-d3a8-44a767f0a12e.png)




### 4-5. 「サマリー」を確認して、「取り込み」をクリック
「サマリー」を確認後、「取り込み」をクリックしてください。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/89ec2d6f-af4c-59e7-80b8-df5404945a90.png)


ジョブが登録されます：
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/4d7855df-e7ce-d319-8f2b-8840f1c15e2d.png)



# 5. 　取り込みジョブの終了確認
状況が「終了」になれば完了です。完了後、データが無事取り込まれたか確認してみてください。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/c5bfb385-6da2-988a-ea06-30e33127c8a6.png)

もし状況が「失敗」の場合はジョブIDをクリックすると、エラーメッセージが表示できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/4b5d3908-3c6b-e2db-99bd-b385feff4d6d.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/e51c5a0f-8c53-dce6-c9c9-4370dde0c6ba.png)

データ取り込みジョブの作成については以上で完了です。

# 6. データの確認
SQLでデータを確認してみましょう!
`lakehouse`は自分のカタログ名、`sensor`は自分のスキーマ名に変更してください。

### 件数
```sql
SELECT COUNT (*) FROM "lakehouse". "sensor". "jpnair"
```
**1106109**件でした。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/5a111067-f29f-aab3-7a75-915d56c02847.png)


### SELECTでLOC_CODE= '01101010' かつ DATA_DATE = '2023/11/01' のデータを取得
```sql
SELECT * FROM "lakehouse"."sensor"."jpnair" WHERE LOC_CODE= '01101010' AND DATA_DATE = '2023/11/01' ;
```
データが表示されました:
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294516/6c8e0fe2-fcc4-fc8d-2695-c6b64d520494.png)

無事データ取り込みできたようですね。
以上です。