BigQueryのテーブル連携時間を監視する - エムスリーテックブログ

これはエムスリー Advent Calendar 2020 の15日目の記事です。前日は id:Hi_king による、臨床AIはなにができ、何が難しいか: 臨床AI研究開発の3類型でした。

エムスリーエンジニアリンググループ AI・機械学習チームの笹川です。趣味はバスケと、筋トレで、このところはNBAのプレシーズンが始まってワクワクしているところです。

今回は、弊社のデータ基盤であるBigQueryへのデータ連携の監視のための便利ツールである tblmonit を開発したので、紹介したいと思います。

github.com

f:id:hsasakawa:20201214100730j:plain — 寒くなってきて、ブランケットにくるまって鼻だけ出してる犬氏（かわいい）

エムスリーのBigQueryの概要
テーブルの更新時間の監視
テーブルメタデータ監視ツール tblmonit
- おまけ
まとめ
We are hiring!

エムスリーのBigQueryの概要

エムスリーではBigQuery上にデータ基盤を構成しており、各種サービスのマスタデータや、ログデータなどが、様々なデータソースから連携され、テーブルとして保存されます。データソースの例としては、

RDB
ログファイル
スプレッドシート
CRMツール

などがあります。

これらを定常的に連携するために、特にdailyでのバッチによるデータ連携についての大部分はDigdagとEmbulkを用いた連携基盤を構築しています。

参考:

www.m3tech.blog

一方で、他の技術スタックから連携されるデータもあり、例えば、機械学習システムが生成した予測値などのデータを直接BQにアップロードしているケースなどがそれにあたります。

テーブルの更新時間の監視

連携されたテーブルに含まれるデータの「正しさ」は、そのデータに依存する施策、意思決定などに多大な影響を与えるという点で、非常に重要な要素です。また、この「正しさ」については、連携されるデータ自体だけでなく、メタデータの監視も特に重要です。今回は特に、データの更新時間について考えてみます。

更新前に、そのデータに依存するジョブが動いてしまうと、古いデータのみを参照してしまうため、ジョブの内部で最新性の確認をしていない限り、一見して正常終了に見えるケースが多いです。このような状況は「間違いに気づくことが難しい」という観点で非常に厄介であるため、更新時間を監視し、遅れなどに気づくことは重要です。

弊社のケースでは、ジョブの成功/失敗、および更新時間に関する通知は、上述のDigdag + Embulkの基盤では、Digdagの機能を使って実現しています。

ジョブエラー通知 Workflow definition — Digdag 0.9.42 documentation
SLA通知 Scheduling workflow — Digdag 0.9.42 documentation

しかしながら、これだけでは十分ではありません。まず、(想像したくありませんが) Digdag + Embulkの基盤自体が動いていないようなケースでは、そもそも上記の通知が行われません。このような問題はcron-job-monitoringと言って、そのためのSaaSがあるような重要な分野です。

また、上述した、Digdag + Embulkを介さずに連携されてくるデータに対しては、そもそも通知機能を使えないため、別の監視手段を考える必要があります。

テーブルメタデータ監視ツール `tblmonit`

上記の問題に対応するため、以下のツールを開発しました。

github.com

このツールでは、はじめに以下のようなtoml形式の設定ファイルを用意します。

[[Project]]
    ID = "bigquery-project-id-1"
    [[Project.Dataset]]
        ID = "dataset1"
        [[Project.Dataset.TableConfig]]
            Table = "table1"
            DateForShards = ""
            Timethreshold = "09:00:00"
            DurationThreshold = "24h"
    [[Project.Dataset]]
        ID = "dataset2"
        [[Project.Dataset.TableConfig]]
            Table = "sharded_table2_on_"
            DateForShards = "ONE_DAY_AGO"
            Timethreshold = "12:00:00"
            DurationThreshold = "1h"
[[Project]]
    Name = "bigquery-project-id-2"
    [[Project.Dataset]]
        ID = "dataset3"
        [[Project.Dataset.TableConfig]]        
            Table = "table1"
            DateForShards = ""
            Timethreshold = "09:00:00"
            DurationThreshold = "24h"

項目の意味を説明すると、以下のようになります。

Project.ID, Dataset.ID, Tableはそれぞれ、GCPプロジェクト、BigQueryデータセットおよび、テーブルを指定する文字列です
DateForShards は、BigQueryでは例えば sample_on_20200101 のようにYYYYMMDD形式のsuffixがつくと、UI上で sample_on_ のprefixを持つテーブルをまとめてくれるのですが、ツールを用いた監視時に対象とするべきsuffixを指定します。TODAY, ONE_DAY_AGO, FIRST_DAY_OF_THE_MONTH に対応しています
DurationThreshold は最新のテーブルが連携されてから、どのくらいの時間がたってもいいか、を設定する値です。 24h と指定した場合、24時間以上経過した場合は、ツールが通知を行います
Timethreshold は、この時間を過ぎても最新テーブルが連携されていない場合は、ツールが通知を行います。これは、 DurationThreshold が前日の連携をイレギュラーな時間に行って (リカバリなどでいつもより遅い時間にもう一度データ連携を行ったなど) 通常の時間との比較が意味をなさない時に利用されます

このファイルをターゲットとして、以下のコマンドを起動します。

tblmonit freshness [target config file]

すると、例えば次のような出力を得ます。

bigquery-project-id-1.dataset1.sharded_table2_on_20200101
bigquery-project-id-2.dataset3.table2

これは、上記の2テーブルが連携されているべき時間に連携されていないという意味です。技術的には、特に凝ったことをしておらず、BigQueryのクライアントを用いて最新のテーブルのlast modified dateを取得し、設定された閾値と見比べて成否を判断しています。

もちろんですが、このジョブは先に記載した「Digdag自体が動いていない問題」に対応するため、Digdagが動いているインフラとは別の環境で動かす必要があります。

おまけ

上記の設定ファイルは、テーブルごとに閾値の設定などを実施する必要がありますが、データセットおよび、その中に含まれるテーブル数が膨大になることが多く、抜け漏れなどのケアが必要です。また、テーブルの連携については、テーブル1つごとの細かな設定が必要なケースより、「このデータセット配下のテーブルはまとめて9時までに連携されていればOK」など、少し緩めの要件にすることが多くありそうです。

そこで、tblmonitには、正規表現などを用いて「ゆるく」設定を書き、それをコンパイルすることで、正しい設定ファイルを出力するサブコマンドを作りました (少し突貫工事なので、いくつかバグがありそうです)。以下は、まさに「このデータセット配下のテーブルはまとめて日次で9時までに連携されていればOK」という設定です。

[[FlexProject]]
    ID = "bigquery-project-id-1"
    [[FlexProject.FlexDataset]]
        ID = "dataset1"
        [[FlexProject.FlexDataset.FlexTableConfig]]
            FlexTable = "*"  # you can use regular expression to specify tables
            DateForShards = "ONE_DAY_AGO"
            Timethreshold = "09:00:00"
            DurationThreshold = "24h"

読んでわかるように、テーブル設定に正規表現を利用できるようにしており、ここでは、あらゆるテーブルにマッチするように * を指定しています。ここは例えば、なんらかのprefixを持つテーブルをまとめたり、特定のテーブルを除外する表現を書くことで柔軟な指定が可能です。データセットに関しても同様な設定が可能になっています。

このファイルをターゲットに次のコマンドを叩きます。

tblmonit config expand [target config file]

すると、例えば、以下の出力を得ます。

[[Project]]
  ID = "bigquery-project-id-1"

  [[Project.Dataset]]
    ID = "dataset1"

    [[Project.Dataset.TableConfig]]
      Table = "table1"
      DateForShards = ""
      TimeThreshold = "09:00:00"
      DurationThreshold = "24h0m0s
    [[Project.Dataset.TableConfig]]
      Table = "sharded_table2_on_"
      DateForShards = "ONE_DAY_AGO"
      TimeThreshold = "09:00:00"
      DurationThreshold = "24h0m0s

テーブルの設定が展開され、 table1 と sharded_table2_on_ の設定が出力されていることがわかります。

当初、この機能は、もとの連携時間監視のコマンドに組み込んで実装したのですが、BigQueryのAPIの制約で、プロジェクト配下のデータセットや、テーブルを全量スキャンする必要があり、実行時間がとんでもなく伸びてしまう問題がありました。そこで、この機能を頻回に実行しない設定出力用の expand コマンドを切り離し、毎回の監視に用いる freshness コマンドはテーブルをexactに照合するだけのシンプルな機能に絞ることにしました。