Amazon ForecastをAWS SDK (Boto3) 経由で操作して時系列予測を作成する

クラスメソッド機械学習 on AWS Advent Calendar 2019

#Amazon Forecast

#機械学習

#AWS

貞松政史

2019.12.04

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

DA事業本部の貞松です。アドベントカレンダーでも引き続きAmazon Forecastです。

本記事は『機械学習 on AWS Advent Calendar 2019』4日目のエントリーです。
クラスメソッド機械学習 on AWS Advent Calendar 2019 - Qiita
クラスメソッド機械学習 on AWS Advent Calendar 2019 ｜シリーズ｜ Developers.IO

これまでマネジメントコンソール上での操作をメインにForecastの機能を解説していましたが、今回はAWS SDK経由で外部のプログラム上から操作する方法について解説していきます。

実行環境

python 3.6.8
boto3 1.10.28
botocore 1.13.28

実際にやってみた

Forecastクライアントの設定

Forecastを操作するためのクライアントを設定します。
認証に必要なアクセスキーとシークレットキーは実際の環境に合わせて設定してください。

import boto3

client = boto3.client(
    'forecast',
    region_name='ap-northeast-1',
    aws_access_key_id = 'xxx',
    aws_secret_access_key = 'xxx',
)

データセットグループの作成

まずはデータセットグループを作成します。
create_dataset_groupに対して、以下のパラメータを渡して実行します。

DatasetGroupName: データセットグループ名
Domain: 予測する時系列データのドメイン
- RETAIL
- CUSTOM
- INVENTORY_PLANNING
- EC2_CAPACITY
- WORK_FORCE
- WEB_TRAFFIC
- METRICS

response = client.create_dataset_group(
    DatasetGroupName='dataset_group_by_sdk',
    Domain='CUSTOM',
)

コンソールで確認すると、ちゃんとデータセットグループが作成されています。

データセットの作成

次にデータセットグループに紐付くデータセットを作成します。
create_dataset_groupに対して、以下のパラメータを渡して実行します。

DatasetName: データセット名
Domain: ドメイン(データセットグループに合わせる)
DatasetType: データセットの種類
- TARGET_TIME_SERIES
- RELATED_TIME_SERIES
- ITEM_METADATA
DataFrequency: 時間の刻み幅
- Y
- M
- W
- D
- H
- 30min
- 15min
- 10min
- 5min
- 1min
Schema: データセットのスキーマ定義

response = client.create_dataset(
    DatasetName='dataset_by_sdk',
    Domain='CUSTOM',
    DatasetType='TARGET_TIME_SERIES',
    DataFrequency='H',
    Schema={
        'Attributes': [
            {
                'AttributeName': 'timestamp',
                'AttributeType': 'timestamp'
            },
            {
                'AttributeName': 'target_value',
                'AttributeType': 'float'
            },
            {
                'AttributeName': 'item_id',
                'AttributeType': 'string'
            }
        ]
    }
)

コンソールを確認すると作成したはずのデータセットがありません。

AWS CLIで確認するとデータセットは作成されているようです。

コンソール上でそうあする場合は、元となるデータセットグループを起点としてそれに紐付くデータセットを作成する為、データセットグループとデータセットの紐付けは自動で行われており特に意識する必要はありませんでした。

SDKでデータセットを作成した場合、データセットグループとデータセットの紐付けを明示的に実行する必要があります。

update_dataset_groupに対して、以下のパラメータを渡して実行することで、データセットグループにデータセットが登録されます。

DatasetGroupArn: データセットグループのArn
DatasetArns: データセットグループに登録するデータセットのArnのリスト

AWS CLIを使用してデータセットグループとデータセットのArnを確認して実行しても良いですが、データセットグループ作成時およびデータセット作成時のレスポンスからArnを取得して保持しておき、それをパラメータとして渡して実行する一連のコードに書き換えてしまった方が簡単そうです。

書き換えたコードは以下のようになります。

create_dataset_group_response = client.create_dataset_group(
    DatasetGroupName='dataset_group_by_sdk',
    Domain='CUSTOM',
)
# 作成したデータセットグループのArnを保持
datasetGroupArn = create_dataset_group_response['DatasetGroupArn']

create_dataset_response = client.create_dataset(
    DatasetName='dataset_by_sdk',
    Domain='CUSTOM',
    DatasetType='TARGET_TIME_SERIES',
    DataFrequency='H',
    Schema={
        'Attributes': [
            {
                'AttributeName': 'timestamp',
                'AttributeType': 'timestamp'
            },
            {
                'AttributeName': 'target_value',
                'AttributeType': 'float'
            },
            {
                'AttributeName': 'item_id',
                'AttributeType': 'string'
            }
        ]
    }
)
# 作成したデータセットのArnを保持
datasetArn = create_dataset_response['DatasetArn']

# データセットグループにデータセットを登録
forecast.update_dataset_group(DatasetGroupArn=datasetGroupArn, DatasetArns=[datasetArn])

今度はちゃんとデータセットグループにデータセットが登録された状態を確認できました。

データセットインポートの作成

データセットの作成が完了したので、次は予測に使用する時系列データをインポートします。データのインポートジョブを作成して実行することでS3バケットに配置されているCSVファイルを読み込んでデータセットにインポートします。

今回は公式のドキュメントでも使用されているUCI Machine Learning リポジトリ(http://archive.ics.uci.edu/ml)で公開されている電力利用量データを用います。

以下のリンクからZIPファイルをダウンロードできます。

electricityusagedata.zip

電力利用量データは以下のような形式となっています。

データカラム
- 利用時間(1時間毎)
- 電力利用量
- クライアントID
データ期間
- 2014-01-01 01:00:00 〜 2015-01-01 00:00:00

create_dataset_import_jobに対して、以下のパラメータを渡して実行することで、データセットインポートジョブを作成、実行します。

response = client.create_dataset_import_job(
    DatasetImportJobName='dataset_import_by_sdk',
    DatasetArn='arn:aws:forecast:ap-northeast-1:xxxxxxxxxxxx:dataset/dataset_by_sdk',
    DataSource={
        'S3Config': {
            'Path': 's3://bucket-name/electricity-usage-data.csv',
            'RoleArn': 'arn:aws:iam::xxxxxxxxxxxx:role/service-role/AmazonForecast-ExecutionRole-xxxxxxxxxxxxx',
        }
    },
    TimestampFormat='yyyy-MM-dd HH:mm:ss'
)

予測子の作成

ようやく予測に使用する時系列データの準備ができたので、予測子を作成します。

create_predictorに対して、以下のパラメータを渡して実行することで、予測子を作成します。

response = client.create_predictor(
    PredictorName='predictor_by_sdk',
    ForecastHorizon=48,
    PerformAutoML=True,
    InputDataConfig={
        'DatasetGroupArn': 'arn:aws:forecast:ap-northeast-1:xxxxxxxxxxxx:dataset-group/dataset_group_by_sdk',
        'SupplementaryFeatures': [
            {
                'Name': 'holiday',
                'Value': 'JP'
            },
        ]
    },
    FeaturizationConfig={
        'ForecastFrequency': 'H'
    }
)

作成した予測子の詳細を確認

予測子の作成が完了したら、予測子の詳細を確認してみます。

get_accuracy_metrics に予測子のARNを渡して実行します。

from pprint import pprint

response = client.get_accuracy_metrics(
    PredictorArn='arn:aws:forecast:ap-northeast-1:xxxxxxxxxxxx:predictor/predictor_by_sdk'
)
pprint(response['PredictorEvaluationResults'])

今回はAutoMLを使用して予測子を作成したので、実行された全てのアルゴリズムとそれらのメトリクスを確認することができます。

ちなみにAutoML使用時に実行された全てのアルゴリズムのメトリクスを取得できるようになったのは直近のアップデートからです。

[アップデート] Amazon ForecastでAutoML使用時に全アルゴリズムのメトリクスを取得できるようになりました

予測の作成

作成済みの予測子を使って予測を作成します。

create_forecastに対して、以下のパラメータを渡して実行することで、予測を作成します。

ちなみにForecastTypesを設定できるようになったのは直近のアップデートからです(５つまで設定可能)

[アップデート] Amazon Forecastで予測作成時に分位数(quantile)を設定できるようになりました

response = client.create_forecast(
    ForecastName='forecast_by_sdk',
    PredictorArn='arn:aws:forecast:ap-northeast-1:xxxxxxxxxxxx:predictor/predictor_by_sdk',
    ForecastTypes=['0.1', '0.3', '0.5', '0.7', '0.9']
)

コンソール上でも無事に予測が作成されていることを確認できました。

予測のエクスポート

予測の作成を実行しただけでは実際の予測値を確認することができないので、まずは予測の結果をエクスポートしてみます。データのエクスポートジョブを作成して実行することで、予測結果のデータをS3バケットにCSV形式でエクスポートします。

create_forecast_export_jobに対して、以下のパラメータを渡して実行することで、予測結果のエクスポートジョブを作成、実行します。

response = client.create_forecast_export_job(
    ForecastExportJobName='forecast_export_by_sdk',
    ForecastArn='arn:aws:forecast:ap-northeast-1:xxxxxxxxxxxx:forecast/forecast_by_sdk',
    Destination={
        'S3Config': {
            'Path': 's3://cm-sadamatsu-test/forecast-export/',
            'RoleArn': 'arn:aws:iam::xxxxxxxxxxxx:role/service-role/AmazonForecast-ExecutionRole-xxxxxxxxxxxxx'
        }
    }
)

コンソール上で予測のエクスポートジョブが作成されていることを確認できます。

エクスポートが完了すると、S3バケット上の指定のパスに予測結果のデータがエクスポートされていることが確認できました。これにより、データを2次加工したり、BIツールに連携したりできます。

予測の可視化

単純に予測結果の状態を確認したい場合は、グラフの描画等により可視化するのが良いでしょう。

予測結果のデータを取り出すには、クライアントから forecastquery サービスに接続して予測結果のデータに対してクエリをかけます。

query_forecastに対して、以下のパラメータを渡して実行することで、予測結果のデータを抽出します。

ForecastArn: 作成済みの予測のArn
Filters: 予測データを絞り込むフィルタ(ここではitem_idで絞り込み)

from pprint import pprint

client = boto3.client(
    'forecastquery',
    region_name='ap-northeast-1',
    aws_access_key_id = 'xxx',
    aws_secret_access_key = 'xxx',
)

response = client.query_forecast(
    ForecastArn='arn:aws:forecast:ap-northeast-1:xxxxxxxxxxxx:forecast/forecast_by_sdk',
    Filters={
        'item_id': 'client_1'
    }
)

pprint(response['Forecast']['Predictions'])

ひとまず予測結果のデータを抽出できました。

抽出したデータを可視化するために今回はmatplotlibを使用します。抽出したデータをnumpy arrayに詰め替えて、matplotlib.pyplotでグラフ化します。

import matplotlib.pyplot as plt
import numpy as np

client = boto3.client(
    'forecastquery',
    region_name='ap-northeast-1',
    aws_access_key_id = 'xxx',
    aws_secret_access_key = 'xxx',
)

response = client.query_forecast(
    ForecastArn='arn:aws:forecast:ap-northeast-1:xxxxxxxxxxxx:forecast/forecast_by_sdk',
    Filters={
        'item_id': 'client_1'
    }
)

x = np.array([x['Timestamp'] for x in response['Forecast']['Predictions']['p10']])
p10y = np.array([p10['Value'] for p10 in response['Forecast']['Predictions']['p10']])
p30y = np.array([p30['Value'] for p30 in response['Forecast']['Predictions']['p30']])
p50y = np.array([p50['Value'] for p50 in response['Forecast']['Predictions']['p50']])
p70y = np.array([p70['Value'] for p70 in response['Forecast']['Predictions']['p70']])
p90y = np.array([p90['Value'] for p90 in response['Forecast']['Predictions']['p90']])

plt.plot(x, p10y, 'blue', label='p10')
plt.plot(x, p30y, 'green', label='p30')
plt.plot(x, p50y, 'yellow', label='p50')
plt.plot(x, p70y, 'orange', label='p70')
plt.plot(x, p90y, 'red', label='p90')

plt.legend()
plt.show()

単純な折れ線グラフですが、p10〜p90までの予測結果をグラフ化することができました。

まとめ

ForecastをAWS SDK経由で外部のプログラム上から操作する方法について解説しました。
最終的にはAPIを呼び出して操作をするだけなので、データセットグループとデータセットの明示的な紐付けにだけ注意すれば、比較的簡単に実装ができそうです。
新規データの投入や再学習、予測のエクスポートや直接クエリをかける処理を工夫して、継続的に時系列予測を作成、利用するフローを組むことにも挑戦したいです。