Google AI Platform - Cloud ML EngineとファイルI/O

ディープラーニングをしていると、やはりGPUリソースを使用したいことが多いわけで。基本を調べて以下の2記事に書きましたが、今回は下記にプラスしてファイル読込・書込について解説します。

- [Google AI Platform - Cloud ML Engineを初心者が動かして理解(前半)](https://qiita.com/FukuharaYohei/items/b9421c2d2837e14eb049)
- [Google AI Platform - Cloud ML Engineを初心者が動かして理解(後半)](https://qiita.com/FukuharaYohei/items/a4621e0a80063047e340)


# 1. ファイル読み書き基本
## 1.1. プログラム
ファイル読み書きはpython標準の`open`ではなくTensorFlowパッケージの[FileIO](https://github.com/tensorflow/tensorflow/blob/master/tensorflow/python/lib/io/file_io.py)を使います。使い勝手はほぼ同じで、`with`構文も使えます。ただ、モードがデフォルトで"r"(読込)にはならないので、明示する必要があります。
パッケージargparseを使って引数にすると、下記のような形で、[GitHub](https://github.com/YoheiFukuhara/gcp-aip-fileio/blob/master/trainer/basic.py)にソース全体を置いています。

```python:basic.py
import argparse
from tensorflow.python.lib.io import file_io

# Parameters
ARGS = None


def main():

    # ローカル/クラウドの両者に対応。モード(r)が必要
    fp = file_io.FileIO(ARGS.input_file, 'r')
    print(fp.read())
    fp.close()

    # with構文もOK
    with file_io.FileIO(ARGS.input_file, 'r') as fp:
        print('with:', fp.read())

    # Write も同じ
    with file_io.FileIO(ARGS.output_file, 'w') as fp:
        fp.write("Hello World!")

    # ローカルに書き込んだらエラーにならないがファイルをあとで受け取れない
    with file_io.FileIO('./ignored.txt', 'w') as fp:
        fp.write("Hello World!")


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '-i', '--input_file',
        default='../input.txt',
        help='Test input data'
    )
    parser.add_argument(
        '-o', '--output_file',
        default='../output.txt',
        help='Test output data'
    )

    ARGS, _ = parser.parse_known_args()
    main()
```

プログラム内で読み込んでいる"input.txt"の内容は以下のようにしています。

```text:input.txt
Input test!!!
```

## 1.2. ローカル実行(Python)

パラメータを指定しないでローカルで動かしてみます(仮想環境を有効にして実行しています)。TensorFlowパッケージの[FileIO](https://github.com/tensorflow/tensorflow/blob/master/tensorflow/python/lib/io/file_io.py)はローカルのディレクトリが指定されれば、ローカルファイルに対して読み書きをしてくれます。

```bash
$ python basic.py
Input test!!!
with: Input test!!!
```

## 1.3. ローカル実行(gcloud ai-platform)

gcloudコマンドを使ってローカルで実行しようとしましたが、`command not found`エラーとなりました。どうもgcloudはPython2.7系に依存しているらしく、私の環境は3.X系だったのでできませんでした。また`unrecognized arguments`エラーも発生しましたが、スペース有無が密接に関連しているようです。スペースを調整したらエラーがでなくなりました([stackoverflow](https://stackoverflow.com/questions/47000556/error-message-while-submitting-google-cloud-ml-training-job-for-tensorflow-objec)参照)。末尾(区切りの"--"部分も)にスペースが必要で、先端(--より前)にはスペース不要です。

```bash
gcloud ai-platform local train \
--module-name trainer.basic \
--package-path trainer \
-- \
--input_file="./input.txt" \
--output_file="./output.txt"
```

省略形のこんな書き方でもOK.

```bash
gcloud ai-platform local train \
--module-name trainer.basic \
--package-path trainer \
-- \
-i="./input.txt" \
-o="./output.txt"
```

## 1.4. クラウド実行
いよいよクラウドで実行します。事前にプロジェクトとバケット作成等をしていま。それらの方法については、別記事[「Google Cloud ML EngineでTensorFlow機械学習訓練実行」](https://qiita.com/FukuharaYohei/items/2a435ef83ecf1702395c)を参考にしてください。


### 1.4.1. 準備：環境変数設定
まずは環境変数にプロジェクト名とバケット名を設定し、ファイルをバケットにアップロードします。

```bash
PROJECT=$(gcloud config get-value project) && BUCKET="${PROJECT}-vcm"
gsutil -m cp ./input.txt gs://${BUCKET}/
```

ジョブ名とパッケージが格納されるディレクトリも環境変数に設定します。"JOB_NAME"はパラメータ"job-dir"に使うのですがハイフン(-)が使用できないことに注意です(詳細は[「Gathering the job configuration data」](https://cloud.google.com/ml-engine/docs/tensorflow/training-jobs#gathering_the_job_configuration_data)参照)。

```bash
now=$(date +"%Y%m%d_%H%M%S")
JOB_NAME="ai_test_$now"
OUTPUT_PATH=gs://$BUCKET/$JOB_NAME
```

### 1.4.2. クラウド実行
gcloudコマンドで実行します。gcloudの標準パラメータと準備した実行するプログラムのパラメータは"-- "で区切ります。スペースが無駄に入っていたり、なかったりで`unrecognized arguments`エラーが出るので注意してください(私は1時間くらい失敗しました・・・)。

```bash
gcloud ai-platform jobs submit training $JOB_NAME \
--job-dir $OUTPUT_PATH \
--module-name trainer.basic \
--package-path trainer/ \
--python-version 3.5 \
--runtime-version 1.13 \
--scale-tier=basic \
-- \
--input_file="gs://$BUCKET/input.txt" \
--output_file="gs://$BUCKET/output.txt"
```

### 1.4.3. ジョブ実行結果確認
gcloudコマンドでログを見ます。

```bash
gcloud ai-platform jobs stream-logs $JOB_NAME
```

プログラム上で読み込みができているのがわかります。

```text:該当箇所ログ
INFO	2019-07-15 16:34:29 +0900	master-replica-0		Input test!!!
INFO	2019-07-15 16:34:29 +0900	master-replica-0		with: Input test!!!
```

### 1.4.4. ストレージ書き込み結果確認
バケット直下のファイルが書き込まれているかを確認します。

```bash
gsutil ls -l  gs://$BUCKET
```

output.txtが作成されているのがわかります。

```bash
        13  2019-07-15T04:34:49Z  gs://gcp-aip-test-vcm/input.txt
       840  2019-07-08T06:04:45Z  gs://gcp-aip-test-vcm/log_config.json
        12  2019-07-15T07:34:29Z  gs://gcp-aip-test-vcm/output.txt
                                 gs://gcp-aip-test-vcm/ai_test_20190715_142132/
                                 gs://gcp-aip-test-vcm/ai_test_20190715_163220/
```

output.txtファイルの中身も正しく書き込まれています。

```bash
$ gsutil cat gs://$BUCKET/output.txt
Hello World!
```

ちなみに下記部分で書いたようなストレージ以外の自身のローカルパスに保存したファイルは取り出しようがないです(あまり調べていないですが、ジョブ実行後にインスタンス消えているはずなので)。ただ、エラーでジョブ終了するわけでないので、重要でないファイル書き込みであれば放置していてもいいかと思います(出力パスが存在しないエラーだけは注意ください)。

```python
    # ローカルに書き込んだらエラーにならないがファイルをあとで受け取れない
    with file_io.FileIO('./ignored.txt', 'w') as fp:
        fp.write("Hello World!")
```

# 2. TensorFlow APIでのファイル書込(Saved ModelエクスポートとTensorBoard)
Python標準の`open`コマンドでなく、Saved Model形式でのモデル保存とTensorBoardのログ出力についてです。[ModelCheckpoint](https://keras.io/ja/callbacks/#modelcheckpoint)を使ったKerasモデルの保存はうまくできませんでした。
KerasでのSaved Model形式でのモデル保存は別記事[「【Keras入門(2)】訓練モデル保存(KerasモデルとSavedModel)」](https://qiita.com/FukuharaYohei/items/ac6333391b8abda94bdc)を参照ください。
KerasでのTensorBoard出力方法については別記事[「【Keras入門(3)】TensorBoardで見える化」](https://qiita.com/FukuharaYohei/items/fe73cdb63ee82a7ea318)を参照ください。

## 2.1. プログラム：モデル保存とTensorBoard出力部分
Save Model形式でのエクスポートとTensorBoard出力部分です。PureなKerasでは試していません。プログラム全体は[GitHub](https://github.com/YoheiFukuhara/gcp-aip-fileio/blob/master/trainer/tf_api.py)に置いています。

```python:tf_api.py
from tensorflow.keras.callbacks import TensorBoard
from tensorflow.contrib import saved_model

# Callbackを定義し、TensorBoard出力の追加
li_cb = []
li_cb.append(TensorBoard(log_dir=ARGS.tensor_board))

# 訓練実行
model.fit(train_x, train_y, epochs=1, callbacks=li_cb)

# Saved Model形式でエクスポート
saved_model.save_keras_model(model, ARGS.saved_model)
```

## 2.2. クラウド実行
Google AI Platformのクラウドで実行します。

```bash
# タイムスタンプを更新してジョブID名をリネーム(同名だとエラー)
now=$(date +"%Y%m%d_%H%M%S")
JOB_NAME="ai_test_$now"
OUTPUT_PATH=gs://$BUCKET/$JOB_NAME
TENSOR_BOARD=gs://$BUCKET/tb
SAVED_PATH=gs://$BUCKET/saved_model

# 実行
gcloud ai-platform jobs submit training $JOB_NAME \
--job-dir $OUTPUT_PATH \
--module-name trainer.tf_api \
--package-path trainer/ \
--python-version 3.5 \
--runtime-version 1.13 \
--scale-tier=basic \
-- \
-t=$TENSOR_BOARD \
-s=$SAVED_PATH \
```

## 2.3. ストレージ書き込み確認
ストレージを見ると書き込みに成功していることがわかります。

```bash
$ gsutil ls -r gs://$BUCKET
gs://gcp-aip-test-vcm/saved_model/:
gs://gcp-aip-test-vcm/saved_model/

gs://gcp-aip-test-vcm/saved_model/1563200064/:
gs://gcp-aip-test-vcm/saved_model/1563200064/
gs://gcp-aip-test-vcm/saved_model/1563200064/saved_model.pb

gs://gcp-aip-test-vcm/saved_model/1563200064/assets/:
gs://gcp-aip-test-vcm/saved_model/1563200064/assets/
gs://gcp-aip-test-vcm/saved_model/1563200064/assets/saved_model.json

gs://gcp-aip-test-vcm/saved_model/1563200064/variables/:
gs://gcp-aip-test-vcm/saved_model/1563200064/variables/
gs://gcp-aip-test-vcm/saved_model/1563200064/variables/checkpoint
gs://gcp-aip-test-vcm/saved_model/1563200064/variables/variables.data-00000-of-00001
gs://gcp-aip-test-vcm/saved_model/1563200064/variables/variables.index

gs://gcp-aip-test-vcm/tb/:
gs://gcp-aip-test-vcm/tb/
gs://gcp-aip-test-vcm/tb/events.out.tfevents.1563200063.cmle-training-1887456742977642267
```

# おまけ
## 使えなくなるAPI
以下は、そこまで時間をかけて調べていませんが、GCSに対して読み書きできなかったAPIです。TensorFlowにラッピングされているKerasを使っても駄目でした。

- [ModelCheckpoint](https://keras.io/ja/callbacks/#modelcheckpoint)を使ったKerasモデルの保存はできませんでした。
- [load_img](https://www.tensorflow.org/api_docs/python/tf/keras/preprocessing/image/load_img)で画像読込ができませんでした。


## loggingパッケージによるログ出力
ファイルI/Oではないですがloggingパッケージを使った場合の出力についてです。loggingパッケージに関しては、別記事[「Pythonでprintを卒業してログ出力をいい感じにする」](https://qiita.com/FukuharaYohei/items/92795107032c8c0bfd19)を参照してください。

### ログ出力サンプルプログラム
loggingパッケージを使って出力しています。パラメータ`-v`が渡されたらDEBUGレベルも出力するようにしています。
プログラム全体は[GitHub](https://github.com/YoheiFukuhara/gcp-aip-fileio/blob/master/trainer/log.py)にあります。

```python:log.py
from logging import getLogger, DEBUG, config
import json
import argparse

from tensorflow.python.lib.io import file_io


# Parameters
ARGS = None


def main():
    with file_io.FileIO(ARGS.log_config, 'r') as f:
        log_conf = json.load(f)

    # replace from INFO to DEBUG if parameter verbose is set
    if ARGS.verbose:
        log_conf["handlers"]["consoleHandler"]["level"] = DEBUG

    # read logging configuration json file
    config.dictConfig(log_conf)

    logger = getLogger(__name__)

    logger.info('Hello World! info')
    logger.debug('Hello World! debug')


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '-l', '--log_config',
        default='../log_config.json',
        help='Log configuration file'
    )
    parser.add_argument(
        '-v', '--verbose', action='store_true',
        help='Log level DEBUG'
    )

    ARGS, _ = parser.parse_known_args()
    main()
```

[ログの設定ファイル](https://github.com/YoheiFukuhara/gcp-aip-fileio/blob/master/log_config.json)は以下のとおりです。

```json:log_config.json
{
    "version": 1,
    "disable_existing_loggers": false,
    "formatters": {
        "default": {
            "format": "%(asctime)s %(name)s:%(lineno)s %(funcName)s [%(levelname)s]: %(message)s"
        }
    },

    "handlers": {
        "consoleHandler": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "default",
            "stream": "ext://sys.stdout"
        }
    },

    "loggers": {
        "__main__": {
            "level": "DEBUG",
            "handlers": ["consoleHandler"],
            "propagate": false
        }
    },

    "root": {
        "level": "INFO"
    }
}
```


### クラウド実行
まずは、log_config.jsonをアップロードします。

```bash
gsutil -m cp ./log_config.json gs://${BUCKET}/
```

あとは、いつもの環境変数設定と実行です。

```bash
# 環境変数設定
now=$(date +"%Y%m%d_%H%M%S")
JOB_NAME="ai_test_$now"
OUTPUT_PATH=gs://$BUCKET/$JOB_NAME
LOG_PATH=gs://$BUCKET/log_config.json

# 実行
gcloud ai-platform jobs submit training $JOB_NAME \
--job-dir $OUTPUT_PATH \
--module-name trainer.log \
--package-path trainer/ \
--python-version 3.5 \
--runtime-version 1.13 \
--scale-tier=basic \
-- \
-l=$LOG_PATH \
-v
```

log上では該当箇所はこんな風に出力されました。AI Platformで実行する場合、タイムスタンプは冗長的なのでformatterから除去してもいいですね。

```bash:該当箇所ログのみ表示
$ gcloud ai-platform jobs stream-logs $JOB_NAME

INFO    2019-07-16 23:15:22 +0900       master-replica-0                2019-07-16 14:15:22,178 __main__:25 main [INFO]: Hello World! info
INFO    2019-07-16 23:15:22 +0900       master-replica-0                2019-07-16 14:15:22,178 __main__:26 main [DEBUG]: Hello World! debug
```