Sparkを利用したOceanBaseの大量データの高速処理および分析

背景

デジタルトランスフォーメーションの波が押し寄せる中、今やデータは企業のコアアセットとなりました。いかに膨大なデータを効率的に処理し、その価値を掘り出すかということが、企業の競争力向上の鍵となっています。OceanBaseデータベースは、高性能な分散データベースであり、トランザクション処理とデータウェアハウスの機能を兼ね備え、大規模データのストレージとクエリをサポートします。一方、Apache Sparkは分散型計算フレームワークであり、その高い計算能力から大量データを処理するための最適なツールです。

本記事では、Sparkを利用してOceanBaseデータベース内の膨大なデータを高速に処理し分析する方法について詳しく解説します。Sparkの分散型計算フレームワークとOceanBaseの高い拡張性を活用することで、データ処理速度とシステムの応答能力を最大限に引き出すことができます。

ビジネスシナリオの紹介

eコマース業界において、ユーザー行動分析は、ビジネスプロセスを最適化し、ユーザーエクスペリエンスを向上させるための主要な手段です。例えば、ユーザーの閲覧や購入などの行動を分析することで、企業はユーザーの嗜好を掘り下げ、潜在顧客を予測し、商品推薦戦略を最適化することができます。

本記事では、eコマースのユーザー行動分析を例として、以下の手順で完全なデータプロセスチェーンを構築する方法について説明します：

1. データの準備：ユーザー行動データセット(CSV形式)を取得します。
2. データストレージ：OceanBaseデータベースにデータをインポートします。
3. データ処理：Spark SQLを利用して、データのクレンジング、変換、および分析を実行します。
4. 結果ストレージ：分析結果をOceanBaseに書き戻します。
5. 詳細な分析：ユーザーの嗜好、人気商品、潜在顧客などの重要な指標について分析します。

データセットフィールドの説明：

• UserID：ユーザーの一意な識別子。
• Timestamp：ユーザー行動が発生した時間。
• PageURL：アクセスするページのURL。
• ProductID：商品の一意な識別子。
• ActionType：ユーザー行動のタイプ(例：browse、cart、purchase)。

本チュートリアルの目標

本チュートリアルでは、以下のユーザー行動分析目標を達成します：

1. ユーザー嗜好のマイニング：プラットフォーム上でのユーザー行動(閲覧、購入など)を分析することで、ユーザーが特定のプロダクトに対して抱く興味や好みを把握します。
2. 潜在顧客の予測：ユーザー行動データを分析することで、購入行為を完了する可能性がある潜在的な優良顧客をスクリーニングします。
3. 人気商品の分析：最近一定の期間に人気があった商品の統計を出し、商品のおすすめを最適化するための根拠とします。

前提条件

開始する前に、以下の環境が整っていることを確認してください：

1. Sparkクラスタ：デプロイと設定が完了しており、Spark SQLを介した分散型計算タスクの実行がサポートされていること。詳細については、Sparkクラスタのデプロイを参照してください。

   また、Sparkインストールパッケージをダウンロードして解凍し、本チュートリアルを試すこともできます。

2. OceanBaseデータベース：デプロイが完了し、JDBCまたはMySQLプロトコルを使用してアクセスできること。MySQL Connector/JによりOceanBaseデータベースに接続するサンプルプログラムを参照してください。

3. データセット：準備済みのユーザー行動CSVファイル(例：user_behavior.csv)は、[https://www.kaggle.com/datasets] からダウンロードできます。

4. 必要な jar パッケージをダウンロードしてインストールします。

   クリックしてspark-connector-oceanbaseをダウンロードします
   クリックしてmysql-connector-jをダウンロードします

   リンクから最新の spark-connector-oceanbaseおよび mysql-connector-jをダウンロードして使用することもできます。

操作手順

ステップ1：SparkとOceanBaseの接続の設定

1. OceanBase接続情報の取得：

   jdbc:mysql://<oceanbase_host>:<port>/<database>?useSSL=false  
   

   • パラメータの説明：

     • oceanbase_host：OceanBaseデータベースのIPアドレスまたはドメイン名。
     • port：ポート番号(ODPリスニングのデフォルトは 2883、直結リスニングのデフォルトは 2881)。
     • database：ターゲットデータベース名。

2. 依存関係のあるドライバーのインストール：

   • spark-connector-oceanbase と mysql-connector-j の jar ファイルをSparkのディレクトリに移動します：

     cp spark-connector-oceanbase-3.4_2.12-1.1.jar $SPARK_HOME/jars/  
     cp mysql-connector-j-8.2.0.jar $SPARK_HOME/jars/  
     

3. Spark Catalogの設定：

   • $SPARK_HOME/conf/spark-defaults.conf を編集し、以下の設定を追加します：

     spark.sql.catalog.ob=com.oceanbase.spark.catalog.OceanBaseCatalog  
     spark.sql.catalog.ob.url=jdbc:mysql://localhost:2881  
     spark.sql.catalog.ob.username=root@test  
     spark.sql.catalog.ob.password=******  
     spark.sql.catalog.ob.schema-name=test  
     

4. Sparkを再起動して、設定を有効にします。

ステップ2：OceanBaseにデータセットを読み込ませる

1. データファイルの準備：

   • サンプルCSVファイル user_behavior.csv の構造：

     UserID,Timestamp,PageURL,ProductID,ActionType  
     U001,2023-01-01 10:00:00,/product/1001,1001,browse  
     U002,2023-01-01 10:15:00,/product/1002,1002,purchase  
     

     Sparkのインストールパッケージをダウンロードして本チュートリアルを体験している場合は、./bin/spark-sql を使用してSpark SQLを起動することができます。

     cd spark-3.x.x-bin-hadoop3
     ./bin/spark-sql
     

     起動に成功すると、一部の出力コードは以下ようになります：

     ...
     Spark master: local[*], Application Id: local-1744093026660
     spark-sql (default)>
     

2. ターゲットテーブルの作成：

   spark-sql (default)>CREATE TABLE user_behavior (  
       UserID STRING,  
       Timestamp TIMESTAMP,  
       PageURL STRING,  
       ProductID STRING,  
       ActionType STRING  
   );  
   

3. データの読み込み：

   • Spark SQLを使用してCSVファイルを読み込み、OceanBaseに書き込む：

     -- CSVファイルを一時ビューに読み込む  
     spark-sql (default)>CREATE OR REPLACE TEMP VIEW user_behavior_data  
     USING csv  
     OPTIONS (  
         path '/path/to/user_behavior.csv',  
         header 'true',  
         inferSchema 'true'  
     );  
     
     -- OceanBaseテーブルに書き込む  
     spark-sql (default)>INSERT INTO user_behavior SELECT * FROM user_behavior_data;  
     

ステップ3：データクレンジングと変換

1. 欠損値の削除：

   spark-sql (default)>SELECT * FROM user_behavior  
   WHERE UserID IS NULL OR Timestamp IS NULL OR ProductID IS NULL OR ActionType IS NULL;  
   

2. 有効な行動のフィルタリング：

   • browse と purchase タイプの動作のみを保持します。

     spark-sql (default)>CREATE OR REPLACE TEMP VIEW cleaned_data AS  
     SELECT *  
     FROM user_behavior  
     WHERE ActionType IN ('browse', 'purchase')  
       AND UserID IS NOT NULL  
       AND ProductID IS NOT NULL;  
     

3. クレンジング結果の検証：

   spark-sql (default)>SELECT COUNT(*) AS total_records FROM cleaned_data;  
   

ステップ4：行動分析と重要指標のマイニング

1. ユーザー行動統計

• ユーザーごとの行動数を集計する：

  spark-sql (default)>SELECT UserID, ProductID, ActionType, COUNT(*) AS ActionCount  
  FROM cleaned_data  
  GROUP BY UserID, ProductID, ActionType  
  ORDER BY UserID, ActionCount DESC;  
  

2. 人気商品の分析

• 商品別の行動数を集計する：

  spark-sql (default)>SELECT ProductID, COUNT(*) AS TotalActions  
  FROM cleaned_data  
  GROUP BY ProductID  
  ORDER BY TotalActions DESC  
  LIMIT 10;  
  

3. 潜在顧客のスクリーニング

• 閲覧回数が5以上のユーザーのスクリーニング：

  spark-sql (default)>SELECT UserID, ProductID, COUNT(*) AS BrowseCount  
  FROM cleaned_data  
  WHERE ActionType = 'browse'  
  GROUP BY UserID, ProductID  
  HAVING BrowseCount > 5;  
  

ステップ5：分析結果をOceanBaseに書き戻す

1. ターゲットテーブルの作成：

   spark-sql (default)>CREATE TABLE potential_users (  
       UserID STRING,  
       ProductID STRING,  
       BrowseCount INT  
   );  
   

2. 潜在顧客データの保存：

   spark-sql (default)>INSERT INTO potential_users  
   SELECT UserID, ProductID, BrowseCount  
   FROM (  
       SELECT UserID, ProductID, COUNT(*) AS BrowseCount  
       FROM cleaned_data  
       WHERE ActionType = 'browse'  
       GROUP BY UserID, ProductID  
       HAVING BrowseCount > 5  
   );  
   

ステップ6：複雑な分析と展開

**1. 行動時間の分布と分析

• 商品閲覧の時間分布の統計：

  spark-sql (default)>SELECT ProductID, HOUR(Timestamp) AS Hour, COUNT(*) AS BrowseCount  
  FROM cleaned_data  
  WHERE ActionType = 'browse'  
  GROUP BY ProductID, HOUR(Timestamp)  
  ORDER BY BrowseCount DESC;  
  

2. ユーザー行動ファネル分析

• ユーザーの閲覧から購入までのコンバージョン率を分析する：

  spark-sql (default)>SELECT UserID,  
         COUNT(CASE WHEN ActionType = 'browse' THEN 1 END) AS BrowseCount,  
         COUNT(CASE WHEN ActionType = 'purchase' THEN 1 END) AS PurchaseCount  
  FROM cleaned_data  
  GROUP BY UserID;  
  

プロセスの概要

Spark + OceanBase の統合ソリューションを通して、以下の完全なプロセスを実現しました：

1. データインポート：Sparkを使用してCSVデータをOceanBaseに読み込みます。
2. データクレンジング：無効なレコードをフィルタリングし、重要な行動データを保持します。
3. 分析とマイニング：
   • ユーザー嗜好分析(行動統計)。
   • 人気商品のマイニング(商品人気度ランキング)。
   • 潜在顧客のスクリーニング(ページビュー数が多いユーザー)。
4. 結果ストレージ：分析結果をOceanBaseに書き戻し、今後の分析に使用します。

プランの価値

1. ストレージとコンピュートの分離：

   • OceanBase は、膨大なデータの効率的なストレージと高同時実行クエリを担当します。
   • Sparkは分散型コンピューティングを担当し、異種データソース間での連合データ分析を実行することで、データ分析の境界を拡大し、分析の効率を向上させます。

2. フルリンクの最適化：

   • データのインポートから結果の出力まで、ETLプロセス全体を上書きします。
   • リアルタイム分析と履歴データマイニングの組み合わせをサポートします。

結論

OceanBaseの高い拡張性を備えたストレージ能力とSparkの分散型コンピューティング能力を活用することで、企業は膨大なユーザー行動データを効率的に処理し、重要なビジネス指標を迅速に分析し、精緻な運用とインテリジェントな意思決定のためのデータサポートを提供することができます。このソリューションは、eコマースシナリオだけでなく、金融、物流など、さまざまな分野にも適用が可能であり、企業のデジタルトランスフォーメーションの重要な技術基盤となります。