【AWS】【Python】Jmespath + フォーマット文字列でフィルタリング

ピンポイントな使い方ですが、Pythonによるjmespathのフィルターについて解説します。

Jmespathとは?

JmespathとはJson形式のドキュメントに対してのクエリ言語です。Jsonの要素を抽出に利用でき、AWS CLI でもフィルターでjmespathが採用されています。 PythonでJmespathを利用する場合、jmespathパッケージをインポートすることで使用できます。

jmespath

Jmespathの基本的な使い方やサンプルは以下のサイトを参考にしていただければ問題ないです。

jmespath.org

ユースケース

例えば、以下のようにAWS SDK for Python 環境下でCWLロググループ/aws/hoge/fuga内のログストリームを取得してログエクスポートする場合、以下のようなコードになります。

import boto3

client = boto3.client('logs')

# ログストリーム取得
log_streams = client.describe_log_streams(logGroupName = "/aws/hoge/fuga",  orderBy = "LastEventTime", descending=True)

# ログエクスポートロジック
for i in log_stream['logStreams']:
     :
     :

上記のlog_streamsの出力結果は以下の通りです。

{
  'logStreams': [
    {
      'logStreamName': '[stream_name_1]', 
      'creationTime': 1622875166781, 
      'firstEventTimestamp': 1622875112000, 
      'lastEventTimestamp': 1625117445000, 
      'lastIngestionTime': 1625118552120, 
      'uploadSequenceToken': '*********************************', 
      'arn': 'arn:aws:logs:[region]:[account_id]:log-group:[log_grp_name]:log-stream:[stream_name_1]', 
      'storedBytes': 0
    }, 
    {
      'logStreamName': '[stream_name_2]', 
        :
    },
      :
  ]
}

上記のロジックでストリームを取得することは可能です。但し、本日分のログを出力したい等のフィルターをかける場合、上記のコードだけではフィルターしきれません。理由として、ログストリームが実行時間ごとに出力されているケースもあり、ストリーム最終更新時のタイムスタンプlastEventTimestampが数週間前のストリームという場合もあるからです。

そのため、jmespathを利用することで、log_streamsの中身を本日分のストリームのみにフィルターして格納する必要があります。

フォーマット文字列による変数宣言を利用したjmespath

当日のログのみを出力するケースで考えていきます。

実行日の0:00:00のエポックタイムを表した変数lastime_epcを定義。各ログストリームのlastEventTimestampがlastime_epc以上の場合、当日更新があったストリームとみなして、log_streamsに格納するロジックに変更したいと思います。条件分岐ではjmespathを利用します。

pythonでjmespathを利用したい場合は、パッケージjmespathをインポートしてjmespath.search('filter', json_data)の形で利用できます。

import jmespath

filterでは、出力するKeyを選択するとともに、JMESPath Examples — JMESPath の通り、要素の条件式として[?Key (演算子) `値` ]とすれば、要素にフィルターをかけることができます。今回の場合はlastEventTimestampがlastime_epc以上とする。

jmespath.search('logStreams[?lastEventTimestamp >= `lastime_epc`].{ 出力したいValue1, ... }', 実行結果)

但し、上記の条件式はテキスト形式で表現する必要があるため変数を直接条件式に挿入することはできない。そのため、Pythonの場合フォーマット文字列を利用して変数の中身を文字列内に記述する必要があります。フォーマット文字列利用時、文字列内で鉤括弧{}を表現する場合は2文字でエスケープ{{}}することをお忘れないように。改良したものは以下のようになる。

jmespath.search('logStreams[?lastEventTimestamp >= `{}` ].{{ 出力したいValue1, ... }}'.format(lastime_epc), 実行結果)

上記のように実施した結果が、以下の通りとなります。

import boto3
import jmespath    # ADD

# Epoch Time 
toda_epc = '1625097600000'    # ADD

client = boto3.client('logs')

# ログストリーム取得
_log_streams = client.describe_log_streams(logGroupName = "/aws/hoge/fuga",  orderBy = "LastEventTime", descending=True)
log_streams = jmespath.search(
            'logStreams[?lastEventTimestamp >= `{}`].{{ logStreamName: logStreamName, lastEventTimestamp: lastEventTimestamp }}'.format(lasttime_epc),
            _log_streams
        )    # ADD

# ログエクスポートロジック
for i in log_stream['logStreams']:
     :
     :

for, if文を利用することなく、シンプルに中身を抽出できました。

終わりに

CLIの時は変数を条件式に定義できたので勝手が違くて最初は迷いましたが、Pythonの基本に立ち返るとなんともないような内容でした。しかし、記載されている文献が意外となかったのでここで残しておきます。

猫キック

IT関連について投稿