リアルタイムマーケティングシステムの紹介とそのリプレイス計画

JBoss Data Grid Java Push

f:id:vasilyjp:20200813134855p:plain

こんにちは、SRE部MA基盤チームの田島です。

私達のチームでは、マーケティングシステムの開発・運用を自前で行っています。マーケティングシステムの内容としては、主にユーザに向けてのメールやLINE・PUSH通知などへの配信です。

マーケティングシステムは大きく分けて2種類あります。1つ目がSQLによるセグメント抽出を行い、抽出したユーザに対してバッチで配信を行うバッチ配信システムです。2つ目がユーザの行動や商品情報等データの変更をリアルタイムに検知して配信を行うリアルタイムマーケティングシステムです。

本記事ではリアルタイムマーケティングシステム（RTM）について紹介します。また現在、RTMのリプレイス計画を行っているのでそれについても紹介いたします。

ZOZOTOWNのリアルタイムマーケティングシステム

リアルタイムマーケティングシステムではユーザの行動や商品情報等データの変更を検知し、ユーザへリアルタイムでアクションを行います。例えばある商品の値段が下がったとき、RTMはその商品情報をリアルタイムに検知し、その商品をお気に入りしているユーザに対して配信を行います。

f:id:vasilyjp:20200813134843p:plain

なぜリアルタイムに配信する必要があるのか

リアルタイム配信を行っている理由は2つあります。1つ目が配信システムのバージョンアップの変遷で、2つ目が配信キャンペーンの性質にあります。

配信システムのバージョンアップの変遷

元々ZOZOTOWNでは1日1回決まった時間に配信をするということしかしていませんでした。

そこでまず、1日に1回抽出したユーザに対して9時・12時・18時の3パターンでユーザ毎に配信時間の振り分けをしてみたしたところ、効果に差が出ました。次に、配信ごとに最新のデータを使ってセグメント抽出し配信をするとCVRの向上がみられました。

このような検証を得てRTMによるリアルタイム配信が誕生しました。リアルタイム配信だけが要因ではありませんが、実際RTMによる配信によって開封率・CTR・CVRの指標が全て向上（最大で開封率2倍、CVR5倍）しました。

配信キャンペーンの性質

ZOZOTOWNのユーザへの通知には、「商品が残り1点になったお気に入りアイテムをユーザにお知らせする」と言ったキャンペーンがあります。そのため、商品が残り1点になり数時間たってからユーザに通知を行っても、既にその商品が完売になっている可能性があります。

また、まだ未実装の状態ではありますが配信チャネルとして訪問中のユーザに対しアプリ内通知としてメッセージを送るということを当初より計画していました。そのため、訪問中のユーザが離脱するよりも前にメッセージを届ける必要があります。

このようなことからZOZOTOWNではリアルタイムマーケティングの仕組みが必要になります。

RTMの機能

イベントの検知から、配信までの流れは以下のようになっています。

f:id:vasilyjp:20200813134828p:plain

以下でそれぞれの機能の詳細を説明します。

イベント検知

まず最初に行うのがイベント検知です。イベントはユーザ行動とZOZOTOWNのデータの変化があります。ユーザ行動はアクセスログから検知を行い、データの変更はZOZOTWONのDBの差分データを取得して検知を行っています。イベント検知されたものを整形しキャンペーン判定の処理に渡します。

キャンペーン判定

イベントを検知するとキャンペーン判定を行います。キャンペーンとはどのようなイベントが発生したときにどのようなお知らせをユーザに提供するかを定めたものです。例えば、商品が値下がりしたときにその商品をお気に入りしているユーザに対して、「あなたがお気に入りしているこの商品が値下がりしました」といったメッセージを配信するキャンペーンがあります。

ユーザ行動や商品情報の変更のタイミングでどのようなキャンペーンを実施するかを決めるのが、このキャンペーン判定の処理になります。

ユーザ抽出/メンバーフィルタ

続いての処理がユーザ抽出処理です。キャンペーンが決まると、そのキャンペーンに該当するユーザの抽出を行います。そしてユーザ抽出後、本当にそのユーザにキャンペーンを配信するべきかのフィルターをユーザごとに行います。これによって得られたユーザに対してキャンペーンの配信を行います。

チャネル最適化

RTMから配信されるチャネルは複数あり、現在MAIL/LINEがあります。また、PUSH配信・アプリ内通知も対象チャネルとして実装を進めています。RTMからどのチャネルに対して配信されるかは、システムによって決められます。例えば、MAILへの反応は悪いがLINEにはよく反応するユーザに対してはLINEのみで配信をするといったことを行います。

時間最適化

RTMではリアルタイム配信を行っていますが、ただただイベント検知からリアルタイムに配信をするだけではなく時間最適化ということも行っています。例えば、このユーザは特定の時間にMAILを確認することが多いと判断すると、リアルタイム配信せず一度キューにためます。その後最適な時間にキューから取り出しキャンペーン配信を行います。もちろん、キャンペーンやチャネルによってリアルタイム配信のほうが良いと判断したら、キューにためずリアルタイム配信を行います。

通数最適化

ユーザへの配信はただ数多く配信をすればいいというものではありません。ユーザが配信をうるさいと感じオプトアウトしてしまうと、ユーザとコミュニケーションを取ることができなくなってしまいます。そこでRTMでは通数最適化を行っています。これはユーザにとって最も最適な通数は何件なのか判断し、それ以上の通知はしないようにしています。

コンテンツ生成・配信処理

コンテンツ生成では、実際に配信するコンテンツを生成します。コンテンツとはLINEであれば、LINEに配信するメッセージと対応するJSONです。

最後に、作成したコンテンツを各チャネルに対し配信を行います。

リアルタイムマーケティングシステムを支える技術

続いて実際にマーケティングシステムで利用している技術について紹介します。

アーキテクチャ

アーキテクチャは以下のようになっています。RTMのシステムはすべてAWS上で動いています。

f:id:vasilyjp:20200813134814p:plain

以下でそれぞれの項目について紹介します。

Analyzer

Analyzerは上記で説明した「キャンペーン判定」から「配信処理」までを行うアプリケーションです。このアプリケーションがRTMのメインアプリケーションとなります。Analyzerについては次の章で詳しく説明します。

RTM DB

RTM DBはRTMのデータを永続化するためのDBです。RTMに関係するデータはこのDBに格納されます。DBにはPostgreSQLを利用しています。

レプリDB

ZOZOTOWNではDBにSQL Serverを利用しており、用途によって複数のSQL Serverに別れています。RTMでは、ZOZOTWONのDBを直接参照するのではなくレプリケーションしたDBをAWS上に立てて利用しています。

SQL Serverではテーブル単位でレプリケーションできるため、各DBから必要なテーブルのみをレプリケーションし、1つのDBに集約しています。

Tracker

TrackerはZOZOTOWNのDBからレプリケーションしたSQL Serverの変更データを追跡し、Analyzerに差分データとして連携するアプリケーションです。差分データの取得にはSQL ServerのChange Trackingという仕組みを利用しています。Change Trackingをかんたんに説明すると、変更が発生したテーブルのPKを保持する機能です。Change Trackingの詳細については以下をご参照ください。

docs.microsoft.com

Change Trackingを有効にすると CHANGETABLE(CHANGES ...) を利用できるようになります。CHANGETABLE(CHANGES ...) からは以下のリンクようなデータを取得できます。

docs.microsoft.com

この機能を利用すると以下のSQLで差分データを取得できます。削除された行に関しては deleted という列を設けて判断できるようにしています。以下はorderテーブルの例です。

SELECT CT.id id,
       O.必要なカラム1,
       O.必要なカラム2
       CASE WHEN O.id IS NULL THEN 1 ELSE 0 END deleted
FROM
CHANGETABLE(CHANGES dbo.order, 前回の取得したときのversion) AS CT
LEFT OUTER JOIN dbo.order AS O ON CT.id = O.id

上記クエリで差分データを取得したらそれをJSONの形式に整形しAnalyzerへ連携します。

Analyzer

先程紹介したとおり、Analyzerが「キャンペーン判定」から「配信処理」までを行うRTMのメインアプリケーションとなっています。リアルタイム配信を実現するため高速かつ柔軟なルール判定が必要となります。それらを実現している技術について紹介します。

Analyzerの非機能要件

「RTMの機能」で説明した機能要件の他に、Analyzerは以下の非機能要件が必須となります。

  • 複雑なルール判定
  • 動的なルール追加
  • 高速な条件判定

複雑なルール判定

Analyzerではユーザの取った様々な行動や、ZOZOTOWNで扱う様々なデータの変化を検知しキャンペーン判定を行います。また各種最適化の処理条件は何パターンにもなり複雑になります。

そこでそれらの複雑なルール判定を統一的かつシンプルに管理できるような仕組みが必要となります。

動的なルール追加

マーケティングシステムのキャンペーン判定や最適化処理のルールはエンジニアではなくビジネスサイドのメンバーによって定められます。そのためプログラムでルール判定をハードコードするのではなく動的にルールの追加や修正できることが必要となります。

高速な条件判定

システムの必須条件としてリアルタイム配信があります。それを実現するためには定義したルール判定を高速に行う必要があります。またメンバーフィルターのような機能はキャンペーン対象のメンバー一人ひとりに対して条件判定を行います。ZOZOTOWNで扱う商品数やユーザ数を考えると同時に発生するキャンペーンの数並びにキャンペーンの対象となるメンバー数は膨大になります。そこでそれらを高速に処理できるよな仕組みが必要となります。

技術構成

上記で紹介してきた機能要件・非機能要件を満たすシステムを構築するためAnalyzerの技術構成は以下のようになっています。

  • 言語
    • Java 8
  • フレームワーク
    • JBoss EAP(Java EE)
    • JBoss Data Grid
    • Red Hat Decision Manager

JBoss EAP

JBoss EAPはRed Hatが提供しているJava EEの実装です。Analyzerでは以下で紹介するJBoss Data GridやRed Hat Decision Managerと組み合わせて機能の実現を行っています。

Red Hat Decision Manager

「複雑なルール判定」並びに「動的なルール追加」を実現するために導入したのがRed Hat Decision Managerです。キャンペーン判定処理やユーザ抽出・メンバーフィルター・チャネル最適化のような処理はすべてルールベースで行っています。それらのルールはすべてRed Hat Decision Managerというルールエンジンによって定義・判定処理を行っています。

Red Hat Decision Mangerを利用することで、Javaで言うところの複雑なswitch文をExeclで表現できます。以下はデバッグユーザ以外を除外するルールを定義したものです（実際に使われているルールではありません）。

f:id:vasilyjp:20200813134703p:plain

3行目がプログラムから渡って来るオブジェクトを定義しており、プログラムからDecision Mangerを呼び出すときに引数で値を渡すことができます。この3行目が実際の条件式になっておりJavaの式をそのまま書くことが可能です。 $param に5行目以降の値が入った状態で評価されます。それぞれオレンジの部分の条件を評価しそれがすべてtrue判定となると一番右のActionが実行されます。

この例では、campaignId == 1 かつ memberId not in (1,2,3) の場合に配信対象外フラグを立てます。campaignIdが2の場合は配信対象外フラグが立つことはありません。また、このExcelで定義したルールはアプリケーションの再起動をすることなしに動的にロードできます。

JBoss Data Grid

Analyzerアプリケーションで1番の肝となる技術がJBoss Data Grid(JDG)です。これによって「高速な条件判定」を実現しています。JDGはかんたんに言うとインメモリな分散キャッシュデータストアで、KeyValue形式でJavaのオブジェクトを直接保存します。JDGでは複数サーバーでクラスタを組み分散してデータを保持します。

JDGにはServer-Client modeとEmbeded modeがありますが、AnalyzerではEmbededモードを利用しています。Embeded modeはJavaアプリケーションとJDGを同一のJVMで動かします。Analyzerでは必要なデータをほぼ全てJDGのキャッシュとして保持しています。このとき以下の機能を組み合わせることですべてのデータへのアクセスを、アプリケーションがアクセスするメモリと同一メモリへのアクセスだけで完結させることが可能です。

  • アプリケーションの分散実行
  • キャッシュが自分のノードに存在するかの判定

これを使った、具体的な処理に関しては次の「メンバー抽出後のメンバーフィルタリング処理」で実例を紹介します。JBoss Data Gridについては以前に以下のブログにて詳しく説明していますので、合わせてご参照ください。

techblog.zozo.com

メンバー抽出後のメンバーフィルタリング処理

どのようにして複雑な条件判定を高速に実現しているかの例として、メンバー抽出後のメンバーフィルタリング処理を擬似コードで紹介します。簡潔にするため、色々と省略しています。

  • メンバーフィルタリングの実行
int campaign_id = xxx;
List<Integer> memberIds = [id1, id2, id3 ...];

AdvancedCache<Object, Object> memberCache = JdgUtil.lookupCache(CacheNames.MEMBER);
DefaultExecutorService des = new DefaultExecutorService(memberCache);

task = new DistributedMemberFilteringService(campaign_id, memberIds);
des.submitEverywhere(task, memberIds)
  • 実際のメンバーフィルタリング処理
public class DistributedMemberFilteringService {
  public DistributedMemberFilteringService(List<Integer> memberIds) {
    this.campaign_id = campaign_id;
    this.memberIds = memberIds;
  }

  public void call() {
    List<Integer> localMemberIds = narrowDownToLocal(memberIds);
    List<Integer> offerMemberIds = new ArrayList<Integer>();

    for(Integer memId : localMemberIds) {
      Member member = (Member)memberCache.get(memberId);
      MemberFilteringFact fact = new MemberFilteringFact(member);
      MemberFilterRule rule = new MemberFilterRule(fact);
      rule.fire();
      if(!fact.isFiltered()) {
        offerMemberIds.add(memberId);
      }
    }
    次の処理(offerMemberIds);
  }

  private List<Integer> narrowDownToLocal(List<Integer> memberIds) {
    List<Integer> localMemberIds = new ArrayList<Integer>();
    AdvancedCache<Object, Object> memberCache = JdgUtil.lookupCache(CacheNames.MEMBER);
    JdgDistributionUtil localityChecker = new JdgDistributionUtil(memberCache);

    for (Integer memberId : memberIds) {
        if (localityChecker.isLocal(memberId)) {
            localMemberIds.add(memberId);
        }
    }
    return localMemberIds;
  }
}

メンバーフィルタリングはメンバー抽出処理によりキャンペーンの対象となりうるメンバーidの一覧が取得できている状態からスタートします。メンバーidはメンバーキャッシュというメンバーの情報を格納しているキャッシュテーブルのkeyとなります。des.submitEverywhere(task, memberIds) を実行すると、memberIdsに含まれるmemberIdをkeyとして1つでも保持しているノード全てでtaskを実行します。

taskは、des.submitEverywhere が呼ばれたタイミングで call() メソッドが実行されます。callメソッドの中で最初に narrowDownToLocal(memberIds) を呼び出しています。JDGの機能に指定したkeyが自分のノードで保持しているかどうかを判定するためのメソッドがあります。それを利用し narrowDownToLocal(memberIds) では、メソッドを実行しているノードが保持しているmemberIdのみになるようmemberIdのListを絞り込みます。これによって、絞り込まれたmemberIdでキャッシュを取得すると必ずアプリケーションと同じメモリからデータを取得できるようになります。

その後、絞り込んだmemberIdのリストをmemberIdごとにRed Hat Dicision Mangerのルールエンジンによってフィルタリングします。ルールの結果はfactのフィールドに保存されます。ルール判定が終わるとfilterに引っかからなかったid(!fact.isFiltered())について次の処理に進みます。

課題

以上のようにリアルタイム配信の仕組みや各種最適化の紹介してきました。それらの機能はうまくいっているものばかりではなく様々な課題があります。現在課題となっているものの一部を紹介します。

リリースの問題

アプリケーションのパフォーマンスを最大化するために、アプリケーションとキャッシュを同一JVMで動作させていると紹介しました。これにより最も課題となるのがリリースになります。

アプリケーションを止めてしまうとそのノードのキャッシュも同時に失われてしまいます。そのためリリースは1台ずつアプリケーションを停止し、キャッシュを別ノードにリバランスしながら行う必要があります。

上記の方法で無停止リリースは可能ですがサーバーが複数台あるため現在3〜4時間ほどリリースに時間がかかっています。メンテナンスタイムを設けることでキャシュをダンプしアプリケーションをリリース、キャッシュをロードし直すといった方法も可能です。この方法であれば1.5時間とリバランスを行うよりも短時間ですみますが、それでも時間がかかります。また、アプリケーションに問題があったときのロールバックにも時間がかかってしまいます。

以上のことからAnalyzerのリリースは慎重に行う必要があり、リリースのサイクルが遅いといった問題があります。

スケーリングの問題

アプリケーションとキャッシュが同一JVMでのっていることはスケーリングも困難にしています。サーバーが増減した場合JDGはキャッシュのリバランスを行います。ただし、サーバーが増えるぶんにはいいのですがサーバーが減る場合適切にキャッシュをリバランスしてやらないとデータロストしてしまいます。またサーバーを減らしすぎた場合キャッシュがクラスター内のメモリにのり切らなくなってしまうと言った問題も発生します。

これらのことから現在RTMではオートスケーリングの仕組み使っておらず常にピークに備えたサーバーを用意して運用しています。

シングルAZ構成

現在AnalyzerシステムはあえてシングルAZ構成にしています。Analyzerで必要となるデータをすべてJDGのキャッシュにのせていると紹介しました。そのためメモリを大量に確保する必要があり、かなり大きめのインスタンス複数台でクラスタを構成しています。

JDGはサーバーが2台以上同時にダウンするとデータロストする設定にしています。そのため、マルチAZ対応を考えるとJDGクラスタをもう一式用意する必要があります。JDGクラスタをもう一式用意するコストと、AZ障害の発生頻度ならびに復旧時間を比較検討した結果シングルAZのほうが好ましいと判断しました。

しかしAZ障害発生時にはダウンタイムは避けられず、ユーザへの配信が滞ってしまいます。これはユーザへの価値提供を犠牲にしていることにほかなりません。

分散キャッシュの運用の難しさ

分散システムは普通のWebアプリケーションよりも格段に複雑で運用が難しくなります。実際に運用しているJDGクラスタがスプリットブレインを起こしたといったことがありました。

techblog.zozo.com

また、JDGの情報がインターネット上に出回っていないため知見があまり得られないという問題もあります。あったとしても数年前の情報ということがほとんどです。このようにAnalyzerを運用し続けることだけでも大変な手間がかかっています。

ルールベースでの条件判定

現在キャンペーン判定やそれぞれの最適化処理はすべてヒューリスティックなルールベースで条件を決めています。ルールはすべてRed Hat Decision Managerで定めており、機械学習での判定処理などはほんの一部にしか適用できていません。

さらにRed Hat Decision Managerの導入理由としてビジネスサイドのメンバーが動的にルールを追加できるようにするためと紹介しました。しかし実際に運用してみるとルールで使うためのデータを参照できるようにする必要があったりと、ルールの追加のたびにプログラムの改修が高頻度で発生してしまっています。

リプレイス計画

以上のような課題を解決するために、RTMシステムの大幅なリプレイスを検討しています。まだ具体的な構成や進め方は決まっていませんが、現在の考えていることを紹介いたします。以下が検討しているアーキテクチャの概要です。

f:id:vasilyjp:20200813134551p:plain

Analyzerのアプリケーションとキャッシュが同じJVMにのっている問題を解決するために何らかのデータストアを外出しすることが第1の目標となります。ただし、外出しすることで現在のパフォーマンスを維持できるかが問題になります。そのため、どのようなデータストアを利用するかと言ったことはまだ決められていません。また、今までAnalyzerの中でやっていた各処理をジョブキューの形にすることで柔軟なスケーリングを実現したいと考えています。

そして現在RTM専用となっていた配信処理の部分を、配信チャネルごとにモジュール化したいと考えています。こうすることでZOZOTOWNのトランザクション処理の中で配信が必要な場合、統一した仕組みで配信できます。

Trackerの部分ですが、現在リアルタイムデータ連携基盤というものを開発しています。それを利用することで差分データを取得できます。アクセスログについても独立した基盤の作成またはツールの導入を検討しています。

まとめ

今回リアルタイムマーケティングシステムについてご紹介しました。紹介したように現在ZOZOTOWNでは、リアルタイムマーケティングシステムのリプレイスを計画しています。この記事をよんでこんなアーキテクチャがいいのでは、もっとこんなことができるのではと思った方はぜひお話しましょう。

tech.zozo.com

また、8/27にリアルタイムマーケティングシステムを含めMAの取り組みについてのイベントを行いますのでぜひご参加ください。

zozotech-inc.connpass.com

カテゴリー