ビジネス・オブザーバビリティ？Elasticでビジネストレンドのレポート自動化に挑戦

はじめに。BIツールで作るレポートの課題。

何かのビジネスのトランザクションのデータに対して分析を行う時、長期的なトレンドを見ながら様々な面に切り分けて観測したいと思います。

BIツールや今回使うKibanaなどを触りながら分析をして、洞察を得ることができるかもしれません。しかし、これはそのツールを使いなれていないと厳しいです。物事を判断する上の人たちがそういったツールを操作することはないですよね。
結局は、BIで作ったデータをパワーポイントなどに貼り付けて報告しているケースが多いんじゃないかと。
しかし、分析する面が増えるほどpptに貼り付けるグラフを作成しなければいけないので、結構大変です。

また、作成したレポートにツッコミを受けるケースがよくあります。

• これってどういう条件で作ったグラフ？
• この数値何かおかしくない？本当？
  そんな時にすぐに集計元のデータを見せながら説明できないと、自分が作った分析で他人を納得させて、アクションを取ってもらうのは大変です。

ElasticsearchとKibanaで課題に挑戦

今回はKibanaを使って、以下を実現してみたいと思います。

• 1ヶ月単位の統計値を過去12ヶ月のデータとともに表示する
• 色々な観点でスライス＆ダイスしたレポートを作る。そして誰でもその結果を簡単に確認できるように、それぞれをPDFとして出力する
• 必要に応じてすぐに集計元データを確認ができるようにする

統計レポートはこんな感じのイメージです。派手な図は必要ありません。。線グラフとその数値がパッと分かれば良いです。
あとはこのレポートを分析したい観点別に提供して欲しいです。

使用する分析データ

Kibanaに付属しているサンプルのFlightデータセットを使います。

出来上がったもの

Webダッシュボード

このように、時系列のグラフと、表で月ごとの統計値を確認できるようにしています。分析する面は国にしました。DestCountry: JP（日本）でフィルタをかけています。
（今回データの準備の都合上、月別ではなく、３日間ごとのデータに分けています）

ダッシュボードのPDF出力

右上のShareのメニューから、Kibanaのレポート機能でPDFとしてエクスポートできます。以下実際のPDFです。Webダッシュボードとほぼ同じ見た目で作成されます。

集計元データの確認画面

上のダッシュボードを見てより詳しくフライト番号などを確認したいとなった場合は、生のフライトデータをKibanaのDiscover機能で確認できます。この画面も、国別のSaved Searchとして保存しておきます。

国ごとにダッシュボードと、集計元データ画面を作成

APIを使ってフィルタする国の文字列だけ変更してコピーをどんどん作成しました。作成にあたってAPIを使いました。

サイトマップ的な各画面へのリンク集

Kibanaを使い慣れていない人でも目的の画面を開けるように、リンク集をダッシュボード上に作りました。

PDFのメール自動送信

ElasticsearchのWatcher機能を使ってダッシュボードをPDFとしてメール送信できます。

どう実装したか

ピボットテーブル

時系列を横軸にした表グラフを作るところをどうするかですが、Kibana LensのTableを使った形式が時系列の集計のピボットテーブルを作成するのに有効でした。Split metrics byで任意の時間感覚に統計を分けていくことができます。

データ別のレポートの作成を自動化

今回の場合、数十個の国分のレポートを手で作るのは大変です。一つ一つGUI上でフィルタしている国を変えて、保存していくこともできますが、これだと手間ですし、運用する上でスケールしません。

KibanaにはダッシュボードやSaved SearchをインポートエクスポートするAPIがあります。
（こちら2022/12/15時点でプレビュー版であり、将来変更になる可能性があることと、正式サポートレベルがこれに関して適用されないことは了承の上、使ってください。）
https://www.elastic.co/guide/en/kibana/8.6/saved-objects-api-export.html

これを活用して、国のところだけを変えたコピーを作っていきます。

今回のコードのサンプルは、pythonのpytestのテストコードとして書いていますが、HTTPを実行できるスクリプトであれば何でも代用可能です。

ダッシュボードをエクスポートするAPIの使用例

import json
import requests
import pytest

def test_export_dashboard():
    kbn_headers = {
        'kbn-xsrf': 'true'
    }
    # ダッシュボードのIDはサンプルです。Kibanaで画面表示した時のURL（以下例）から確認できます。
    # https://<kibana_endpoint>/app/dashboards#/view/05dded0e-7b43-49d6-85a6-7389e3c11108?_g=(filters:!(),refreshInterval:(pause:!t,value:0),time:(from:now-30d%2Fd,to:now))

    source_dashboard_id = "05dded0e-7b43-49d6-85a6-7389e3c11108"
    data = json.dumps(
        {
            "objects": [
                {
                "type": "dashboard",
                "id": f"{source_dashboard_id}"
                }
            ]
        }
    )
    session = requests.Session()
    session.auth = ('<user>', '<password>')
    r = session.post(f'https://<kibana_endpoint>/api/saved_objects/_export', data=data, headers=kbn_headers)

    open(f'Flight by Country [JP].ndjson', 'w').write(r.text)
    assert r.status_code == 200

エクスポートしたダッシュボードはndjson形式のファイルです。国相当分ファイルをコピーし、Filterのquery部分に該当する文字列"JP"を別のフィルターしたい文字列に置換します。

import json
import pytest

def test_clone_dashboasrds():
    countries = ['IT','US','CN','CA','JP','RU','CH','GB','AU','PL','AT','IN','AR','ZA','DE','SE','EC','KR','NO','PR','MX','CO','PE','FI','DK','CL','AE','IE','FR','ES','TR','NL']
    with open('Flight by Country [JP].ndjson', 'r') as f:
        contents = f.read()

    for country in countries:
        new_contents = contents.replace('JP', country)
        open(f'Flight by Country [{country}].ndjson', 'w').write(new_contents)

作成した国別のダッシュボードndjsonファイルをインポートします。

import json
import requests
import pytest

def test_import_dashboard():
    kbn_headers = {
        'kbn-xsrf': 'true'
    }
    countries = ['IT','US','CN','CA','JP','RU','CH','GB','AU','PL','AT','IN','AR','ZA','DE','SE','EC','KR','NO','PR','MX','CO','PE','FI','DK','CL','AE','IE','FR','ES','TR','NL']
    for country in countries:
        files = {'file': open(f'Flight by Country [{country}].ndjson', 'r')}

        session = requests.Session()
        session.auth = ('<user>', '<password>')
        r = session.post(f'https://<kibana_endpoint>/api/saved_objects/_import?createNewCopies=true', files=files, headers=kbn_headers)
        assert r.status_code == 200

Saved SearchのExport/Importも同じやりかたです。エクスポートしたファイルのコピーも、フィルタリングしている文字列だけすり替えてインポートです。

import json
import pytest

def test_clone_savedsearch():
    countries = ['IT','US','CN','CA','JP','RU','CH','GB','AU','PL','AT','IN','AR','ZA','DE','SE','EC','KR','NO','PR','MX','CO','PE','FI','DK','CL','AE','IE','FR','ES','TR','NL']
    with open('Saved Search - Flight by Country [JP].ndjson', 'r') as f:
        contents = f.read()

    for country in countries:
        new_contents = contents.replace('JP', country)
        open(f'Saved Search - Flight by Country [{country}].ndjson', 'w').write(new_contents)

こちらはサイトマップみたいなダッシュボードを作るコードです。
ダッシュボードに一つMarkdownのテキストのビジュアルを入れておき、エクスポートしてください。
それに対して、Dashboardそれぞれの静的webリンクを生成して、markdownに追加しています。

import json
import requests
import pytest

kbn_headers = {
    'kbn-xsrf': 'true'
}

def test_create_custom_index_page_of_objects():
    session = requests.Session()
    session.auth = ('<user>', '<password>')
    r = session.get(f'https://<kibana_endpoint>/api/saved_objects/_find?type=search&search_fields=title&search="Saved Search - Flight by Country"*', headers=kbn_headers)
    assert r.status_code == 200

    markdown_lines = []
    markdown_lines.append('# データへのリンク\\\\n---')
    for object in json.loads(r.text)['saved_objects']:
        markdown_lines.append('[%s](https://<kibana_endpoint>/app/discover#/view/%s)' % (object['attributes']['title'], object['id']))

    with open('linkpage_blank.ndjson', 'r') as f:
        contents = f.read()

    markdown = '\\\\n\\\\n'.join(markdown_lines)
    contents = contents.replace('FILL', markdown, )
    with open('linkpage.ndjson', 'w') as f:
        f.write(contents)

def test_import_custom_index_page_of_objects():
    session = requests.Session()
    session.auth = ('<user>', '<password>')
    files = {'file': open('linkpage.ndjson', 'r')}
    r = session.post(f'https://<kibana_endpoint>/api/saved_objects/_import?overwrite=true', files=files, headers=kbn_headers)
    assert r.status_code == 200

WatcherでPDFを自動メール送信するところです。
こちらのドキュメントを参考にしています。
https://www.elastic.co/guide/en/kibana/8.6/automating-report-generation.html

また、前提として、API経由でPDFレポートを作成するためには、別途以下の設定が必要です。こちらを事前に実施しておきます。
https://www.elastic.co/guide/en/kibana/8.6/secure-reporting.html

import json
import requests
import time
import pprint
import pytest

kbn_headers = {
    'kbn-xsrf': 'true'
}

def test_watcher_pdf_report():
    session = requests.Session()
    session.auth = ('<user>', '<password>')
    countries = ['IT','US','CN','CA','JP','RU','CH','GB','AU','PL','AT','IN','AR','ZA','DE','SE','EC','KR','NO','PR','MX','CO','PE','FI','DK','CL','AE','IE','FR','ES','TR','NL']

    
    # 以下は上でインポートして作成したサイトマップのダッシュボードへのリンクです。適宜置き換えてください。
    sitemap_dashboard_url = 'https://<kibana_endpoint>/app/dashboards#/view/f7975e64-a44c-450c-8e79-052b735bf6c1'

    for country in countries:
        r = session.get(f'https://<kibana_endpoint>/api/saved_objects/_find?type=dashboard&search_fields=title&search="Flight Trends by City [{country}]"', headers=kbn_headers)
        assert r.status_code == 200

        #　PDFにしたいダッシュボードのIDを取得しています。下の"url"の中で使用します。
        id = json.loads(r.text)['saved_objects'][0]['id']

        data = json.dumps(
            {
            "trigger" : {
                "schedule": {
                "interval": "1d"
                }
            },
            "actions" : {
                "email_admin" : { 
                "email": {
                    "to": "'Recipient Name <[e-mail]>'",
                    "subject": f"フライト日次レポート[{country}]",
                    "body" : f"国[{country}]到着便の集計レポートについて添付ファイルをご確認ください。\n また、詳細なフライトデータに関してはこちらをご確認ください。\n {sitemap_dashboard_url} ",
                    "attachments" : {
                    f"flight_report_{country}.pdf" : {
                        "reporting" : {
                        "url": f"https://<kibana_endpoint>api/reporting/generate/printablePdfV2?jobParams=%28browserTimezone%3AAsia%2FTokyo%2Clayout%3A%28dimensions%3A%28height%3A1071.99658203125%2Cwidth%3A2844.444580078125%29%2Cid%3Apreserve_layout%29%2ClocatorParams%3A%21%28%28id%3ADASHBOARD_APP_LOCATOR%2Cparams%3A%28dashboardId%3A%27{id}%27%2CpreserveSavedFilters%3A%21t%2CtimeRange%3A%28from%3Anow-30d%2Fd%2Cto%3Anow%29%2CuseHash%3A%21f%2CviewMode%3Aview%29%29%29%2CobjectType%3Adashboard%2Ctitle%3A%27Flight%20Trends%20by%20City%20%5BJP%5D%27%2Cversion%3A%278.5.0%27%29", 
                        "retries":40, 
                        "interval":"15s", 
                        "auth":{ 
                            "basic":{
                            "username":"<user>",
                            "password":"<password>"
                            }
                        }
                        }
                    }
                    }
                }
                }
            }
            }
        )

        r = session.put(f'{PROTOCOL}://{ES_HOST}:{ES_PORT}/_watcher/watch/flight_report_{country}', data=data, headers=headers)
        pprint.pprint(r.json())
        assert r.status_code == 201 or 200, r.json()['error']['reason']

まとめ

最後ちょっとスクリプトが必要ですが、KibanaのAPIやPDF出力があるお陰で、たくさんのKibanaレポートを自動作成して、それを配布する運用が実現できると思います。
なお、WatcherとPDF出力の利用はオンプレダウンロード版の場合、プラチナの有償ライセンスが必要となります！