Apache Cassandra と Apache Pulsar を使用した小売業への AI の実装

人工知能と機械学習のソリューションを実装するには、デジタル システムで日常的に発生する多くの一般的な課題を解決する必要があります。レガシー システムの更新、バッチ プロセスの排除、AI/ML に基づいた革新的なテクノロジーの使用による顧客エクスペリエンスの向上などです。ほんの数年前にはSFのように思えた方法です。

この進化を説明するために、大手小売店で AI/ML ソリューションの実装を支援するために雇われた架空の請負業者を追跡してみましょう。これは、AI/ML への取り組みの重要な側面を詳しく説明する一連の記事の最初の記事です。

問題: 脆弱なバッチプロセス

今日は BigBoxCo で「インフラストラクチャ」チームの初日です。義務付けられている人事活動を終えた後、私は請負業者バッジを受け取り、新しい職場へ向かいました。

チームに会った後、今朝「推薦」チームと会議があると言われました。

私のシステムへのアクセスはまだ十分に機能していないので、会議中に IT 部門が問題を解決してくれることを願っています。

会議室には、私のマネージャーと新しいチームのエンジニア 2 人、レコメンデーション チームのエンジニア 1 人の数人だけです。まずはいくつかの紹介から始めて、その後、前の週の問題について話し合っていきます。

明らかに、先週、ある種の夜間のバッチ障害が発生し、彼らはまだその影響を感じています。

現在の製品の推奨は、顧客の注文から収集されたデータに基づいているようです。注文ごとに、注文された製品間に新しい関連付けが作成され、記録されます。

顧客が製品ページを表示すると、現在の製品を別の製品と一緒に購入した他の顧客の数に基づいて推奨事項を得ることができます。

製品の推奨事項は、クラウドのマイクロサービス層を介して bigboxco.com のユーザーに提供されます。マイクロサービス層は、 Apache Cassandraのローカル (クラウド) データセンター デプロイメントを使用して結果を提供します。

ただし、結果がどのように収集されて提供されるかは、まったく別の話になります。基本的に、(一緒に購入した) 製品間の関連付けの結果は、MapReduce ジョブ中にコンパイルされます。これは先週失敗したバッチ プロセスです。

このバッチ プロセスは決して高速ではありませんでしたが、時間の経過とともに遅くなり、脆弱になってきました。実際、プロセスの実行に 2 日、場合によっては 3 日かかる場合もあります。

エクスペリエンスの向上

会議が終わってパソコンを確認すると、ようやくログインできたようです。辺りを見回していると、プリンシパルエンジニア（PE）がやって来て自己紹介をします。

私が彼に Recommendations チームとのミーティングについて話すと、彼は Recommendation サービスの背後にある歴史についてもう少し詳しく教えてくれました。

このバッチプロセスは約 10 年前から導入されているようです。それを設計したエンジニアは退職し、組織内でそれを本当に理解している人はほとんどおらず、誰も触ろうとしません。

もう 1 つの問題は、各推奨事項を決定するデータセットがほとんどの場合、数日前のものであることです。

これは大局的には大したことではないかもしれませんが、推奨データをより最新のものにすることができれば、マーケティング部門が実施する短期的なプロモーションに利益をもたらすでしょう。

彼は同意してうなずき、システムを改善する方法についての提案を歓迎すると言いました。

もしかしたらグラフの問題なのでしょうか？

最初は、これはグラフの問題のように思えます。サイトにログインして商品を購入する顧客もいます。その前に、ユーザーが商品を見たり、カートに追加したりしたときに、「X を購入した顧客は Y も購入しています」という形で推奨事項を表示できます。

現在このサイトには、レコメンデーション サービスがまさにこれを実行する機能があり、頻繁に一緒に購入される上位 4 つの追加製品が返されます。

しかし、2 億人の顧客のいずれかが同時に購入する 1 つの製品と他のすべての製品とのマッピングは、急速に大規模になるため、製品を「ランク付け」する何らかの方法が必要になります。

したがって、注文内に表示される回数によってそれらをランク付けできます。このシステムのグラフは、以下の図 1 のようになります。

これをモデル化し、実際のデータ量を使用してグラフ データベース上で実行した後、これではうまくいかないことがすぐにわかりました。

1 つの製品から近くの顧客、さらにその製品をたどって、最も多く表示される製品を計算するのにかかる時間は約 10 秒です。

基本的に、私たちは 2 日間のバッチ問題を「パント」し、各ルックアップでトラバーサル レイテンシを望まない場所、つまり顧客の目の前に正確に配置するようにしました。

しかし、おそらく、そのグラフ モデルは、ここで行う必要があることからそれほど遠くないのではないでしょうか?実際、上記のアプローチは、「協調フィルタリング」として知られる機械学習 (ML) 技術です。

基本的に、協調フィルタリングは、他のユーザーとのアクティビティに基づいて特定のデータ オブジェクトの類似性を調べるアプローチであり、そのデータに基づいて予測を行うことができます。

私たちの場合、顧客ベースからカート/注文データを暗黙的に収集し、それを使用してより適切な製品推奨を行い、オンライン売上を増やします。

実装

まずはデータ収集から見ていきましょう。ショッピングの「注文」機能に追加のサービス呼び出しを追加することは、それほど大したことではありません。実際、それはすでに存在します。データがデータベースに保存され、後で処理されるだけです。間違いなく、バッチ処理を含めたいと考えています。

ただし、そのカート データをリアルタイムで処理して、オンライン データ セットに直接フィードバックして、その後すぐに使用できるようにしたいとも考えています。

まずは、次のようなイベント ストリーミング ソリューションを導入します。アパッチパルサー。そうすることで、すべての新しいカート アクティビティがパルサーの話題、そこで消費され、基礎となるバッチ データベースの両方に送信されるだけでなく、リアルタイム ML モデルのトレーニングにも役立ちます。

後者に関しては、Pulsar のコンシューマーは、注文内の各製品のエントリを保持するためだけに設計された Cassandra テーブル (図 2 を参照) に書き込みます。この製品には、その注文と他の注文からの他のすべての製品の行が含まれます。

 CREATE TABLE order_products_mapping ( id text, added_product_id text, cart_id uuid, qty int, PRIMARY KEY (id, added_product_id, cart_id) ) WITH CLUSTERING ORDER BY (added_product_id ASC, cart_id ASC); 

次に、次のように、特定の製品 (この例では「DSH915」) についてこのテーブルをクエリできます。

 SELECT added_product_id, SUM(qty) FROm order_products_mapping WHERE id='DSH915' GROUP BY added_product_id; added_product_id | system.sum(qty) ------------------+----------------- APC30 | 7 ECJ112 | 1 LN355 | 2 LS534 | 4 RCE857 | 3 RSH2112 | 5 TSD925 | 1 (7 rows)

次に、上位 4 つの結果を取得して製品推奨テーブルに入力し、推奨サービスが「 product_id 」でクエリできるようにします。

 SELECT * FROM product_recommendations WHERE product_id='DSH915'; product_id | tier | recommended_id | score ------------+------+----------------+------- DSH915 | 1 | APC30 | 7 DSH915 | 2 | RSH2112 | 5 DSH915 | 3 | LS534 | 4 DSH915 | 4 | RCE857 | 3 (4 rows)

このようにして、新しい推奨データは常に最新の状態に保たれます。また、上記のインフラストラクチャ資産はすべてローカル データ センターにあります。

したがって、注文から製品関係を取得し、Pulsar トピックを通じて送信し、Cassandra に保存された推奨事項に処理するプロセスは 1 秒以内に実行されます。

この単純なデータ モデルを使用すると、Cassandra は要求された推奨事項を 1 桁のミリ秒で提供できます。

結論と次のステップ

長期的には、データがどのように Cassandra テーブルに書き込まれているかを必ず調査したいと考えています。このようにして、バインドされていない行の増加やインプレース更新などに関連する潜在的な問題を事前に解決できます。

「推奨しない」リストなど、追加のヒューリスティック フィルターも追加する必要がある場合があります。

これは、顧客が一度またはまれに購入する製品がいくつかあり、それらを推奨しても、衝動的に購入する可能性がはるかに高い他の製品のスペースを奪うだけだからです。

たとえば、洗濯機などの家電部門の商品の購入を勧めても、「衝動買い」が起こる可能性は低いです。

将来のもう 1 つの改善点は、Kaskadaのようなリアルタイム AI/ML プラットフォームを実装して、製品関係ストリーミングを処理し、レコメンデーション データをサービスに直接提供することです。

幸いなことに、私たちは Pulsar を使用して、リアルタイムで処理されるカート追加イベントをフィードすることで、既存の遅いバッチ プロセスを強化する方法を思いつきました。このシステムが長期的にどのように機能するかを把握したら、従来のバッチ プロセスをシャットダウンすることを検討する必要があります。

PE は、新しいソリューションで順調に進歩したこと、そしてさらに良いことに、技術的負債の一部を解消するための基礎を築き始めたことを認めました。結局のところ、誰もがそれで良いと感じます。

今後の記事では、 ベクトル検索を使用した製品プロモーションの改善について見ていきます。

DataStax がリアルタイム AI をどのように実現するかを学ぶ

DataStax、Aaron Ploetz 著

ここでも公開されています