OB Cloudを活用したオフラインデータウェアハウスと可視化ダッシュボードの構築

背景情報

社会のデジタルトランスフォーメーションが加速するにつれて、企業のデータ分析に対する要求は日々高まっています。データ処理は主に大量の履歴データを処理するオフラインシナリオに集中しており、データウェアハウスを構築することで解決できます。OB Cloud は、データウェアハウスを構築することでオフラインデータの分析問題を解決できます。スケジュールされたタスクを通じて、オペレーショナルデータストア (ODS) から詳細データストア (DWD)、そして最終的なアプリケーションデータストア (ADS) までのデータ階層構築を実現し、OB Cloud ベースのデータウェアハウスを作成します。同時に、関連するエコシステムツールを使用して可視化ダッシュボードを構築します。

前提条件

• 現在のインスタンスのインスタンス管理者およびデータ読み取り権限を持っていること。権限がない場合は、組織管理者に連絡して追加を依頼してください。
• 利用可能な分析型（MySQL）クラスタインスタンスまたは共有インスタンスがあること。クラスタインスタンスの場合はテナントの作成を参照してテナントを作成してください。現在利用可能なインスタンスがない場合は、OB Cloud クラウドデータベースの無料トライアルインスタンスを開通できます。無料トライアルのルールと開通方法をクリックして確認してください。
• 使用しているデータベースのカーネルバージョンが 4.3.0 より新しく、ソース側で Binlog サービスが有効になっていること。
• 組織管理者権限を持ち、クラスタインスタンスを作成済みであること。詳細はクラスタインスタンスの作成を参照してください。
• プロジェクト内のインスタンスを操作できるプロジェクト管理者またはインスタンス管理者であること。権限がない場合は、組織管理者に連絡して権限を追加してください。
• Flink CDC、DBT、Airflow、Grafana、Prometheus をデプロイ済みであること。
• Prometheus による監視メトリクスの収集が完了し、Grafana と Prometheus 間のネットワーク通信が可能であること。また、OB Cloud 監視ダッシュボードの設定ファイルをダウンロード済みであること：ホスト監視 および インスタンス監視。これら2つのファイルは事前に設定されたダッシュボードのサンプルファイルです。

準備作業

アカウントの作成

1. tutorial_tenant テナントを選択します。

2. テナントコンソールページにアクセスします。
3. アカウントの作成 をクリックします。
4. ポップアップウィンドウでアカウント名に tutorial_user を入力します。テナントアカウント名はカスタマイズ可能です。
5. アカウントタイプとして一般アカウントを選択します。
6. アイコンをクリックし、権限を付与するデータベースとして default_database を選択します。

7. アイコンをクリックし、現在のアカウントに読み書き権限を付与します。

8. 「ランダム生成」をクリックします。クリックするとシステムが自動的にパスワードを生成するので、コピーして安全に保管してください。パスワードをカスタマイズすることも可能です。パスワードの長さは 10～32 文字で、大文字英字 2 文字以上、小文字英字 2 文字以上、数字 2 文字以上、および特殊文字 2 文字以上を含める必要があります。サポートされる特殊文字は ~ ! @ # % ^ & * _ - + = \| ( ) { } [ ] : ; , . ? / | です。

9. 作成 をクリックします。

データベースの作成とテーブルの構築

1. テナントコンソール ページにアクセスします。
2. データベースの作成 をクリックし、ポップアップウィンドウにデータベース名 tptest を入力して文字セットを選択し、TP データベースとして設定します。
3. 作成 をクリックします。
4. 手順 1～3 を繰り返し、AP データベースとして aptest データベースを作成します。
5. TP データベースと AP データベースのそれぞれにテーブルを作成します。

   1. SQL コンソール ページにアクセスします。

   2. tutorial_user アカウントを選択してパスワードを入力してログインし、確認 をクリックします。

   3. 左側の tptest データベースをダブルクリックし、新しい SQL ウィンドウを開きます。

   4. 以下の SQL を実行します。

      CREATE TABLE `orders` (
      order_id bigint not null primary key,
      user_id varchar(50) not null,
      shop_id bigint not null,
      product_id bigint not null,
      buy_fee numeric(20,2) not null,   
      create_time timestamp not null,
      update_time timestamp not null default now(),
      state int not null 
      );
      
      CREATE TABLE `orders_pay` (
      pay_id bigint not null primary key,
      order_id bigint not null,
      pay_platform varchar(64) not null,
      create_time timestamp not null
      );
      
      CREATE TABLE `product_catalog` (
      product_id bigint not null primary key,
      catalog_name varchar(50) not null
      );
      

   5. アイコンをクリックして SQL 文を実行します。実行履歴 タブに作成されたテーブル情報が表示されます。

   6. 上記の手順 3～5 を繰り返し、aptest データベースに同じ 3 つのテーブルを作成します。

   7. 左側の tptest データベースをダブルクリックし、再度新しい SQL ウィンドウを開きます。

   8. 以下の SQL を実行し、tptest データベースの 3 つのテーブルに模擬データをいくつか挿入します。

      INSERT INTO product_catalog VALUES(1, 'iphone 14'),(2, 'iphone 14 pro max'),(3, 'iphone 15'),(4, 'huawei mate 60'),(5, 'huawei pura 70');
      
      insert into `tptest`.`orders_pay`(`pay_id`,`order_id`,`pay_platform`,`create_time`) values(1,1,'test','2024-10-01 00:00:00'); 
      insert into `tptest`.`orders_pay`(`pay_id`,`order_id`,`pay_platform`,`create_time`) values(2,2,'test','2024-10-02 00:00:00'); 
      insert into `tptest`.`orders_pay`(`pay_id`,`order_id`,`pay_platform`,`create_time`) values(3,3,'test','2024-10-03 00:00:00'); 
      insert into `tptest`.`orders_pay`(`pay_id`,`order_id`,`pay_platform`,`create_time`) values(4,4,'test','2024-10-04 00:00:00');
      
      insert into `tptest`.`orders_pay`(`pay_id`,`order_id`,`pay_platform`,`create_time`) values(1,1,'test','2024-10-01 00:00:00'); 
      insert into `tptest`.`orders_pay`(`pay_id`,`order_id`,`pay_platform`,`create_time`) values(2,2,'test','2024-10-02 00:00:00'); 
      insert into `tptest`.`orders_pay`(`pay_id`,`order_id`,`pay_platform`,`create_time`) values(3,3,'test','2024-10-03 00:00:00'); 
      insert into `tptest`.`orders_pay`(`pay_id`,`order_id`,`pay_platform`,`create_time`) values(4,4,'test','2024-10-04 00:00:00');
      

   9. アイコンをクリックして SQL 文を実行します。実行履歴 タブに関連情報が表示されます。

Flink 同期リンクの構築

Flink CDC を起動後、Flink をデプロイしたフォルダに入り、./bin/sql-client.sh コマンドを実行して Flink SQL インターフェースを開きます。以下の SQL を順に実行します。

```sql
CREATE TABLE mysql_orders (
order_id bigint not null primary key NOT ENFORCED,
user_id varchar(50) not null,
shop_id bigint not null,
product_id bigint not null,
buy_fee numeric(20,2) not null,   
create_time timestamp not null,
update_time timestamp not null,
state int not null 
) WITH (
    'connector' = 'mysql-cdc',
    'server-time-zone' = 'Asia/Shanghai',
    'hostname' = '******.huawei-cn-southwest-2.oceanbase.cloud',
    'port' = '3306',
    'username' = 'wktest',
    'password' = '******',
    'database-name' = 'tptest',
    'table-name' = 'orders');

CREATE TABLE `mysql_orders_pay` (
pay_id bigint not null primary key NOT ENFORCED,
order_id bigint not null,
pay_platform varchar(64) not null,
create_time timestamp not null
) WITH (
    'connector' = 'mysql-cdc',
    'server-time-zone' = 'Asia/Shanghai',
    'hostname' = '******.huawei-cn-southwest-2.oceanbase.cloud',
    'port' = '3306',
    'username' = 'wktest',
    'password' = '******',
    'database-name' = 'tptest',
    'table-name' = 'orders_pay');

CREATE TABLE `mysql_product_catalog` (
product_id bigint not null primary key NOT ENFORCED,
catalog_name varchar(50) not null
) WITH (
    'connector' = 'mysql-cdc',
    'server-time-zone' = 'Asia/Shanghai',
    'hostname' = '******.huawei-cn-southwest-2.oceanbase.cloud',
    'port' = '3306',
    'username' = 'wktest',
    'password' = '******',
    'database-name' = 'tptest',
    'table-name' = 'product_catalog');

CREATE TABLE `orders` (
order_id bigint not null primary key NOT ENFORCED,
user_id varchar(50) not null,
shop_id bigint not null,
product_id bigint not null,
buy_fee numeric(20,2) not null,   
create_time timestamp not null,
update_time timestamp not null,
state int not null 
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://******.huawei-cn-southwest-2.oceanbase.cloud:3306/aptest',
    'username' = 'wktest', 
    'password' = '******',
    'table-name' = 'orders');

CREATE TABLE `orders_pay` (
pay_id bigint not null primary key NOT ENFORCED,
order_id bigint not null,
pay_platform varchar(64) not null,
create_time timestamp not null
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://******.huawei-cn-southwest-2.oceanbase.cloud:3306/aptest',
    'username' = 'wktest', 
    'password' = '******',
    'table-name' = 'orders_pay');

CREATE TABLE `product_catalog` (
product_id bigint not null primary key NOT ENFORCED,
catalog_name varchar(50) not null
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://******.huawei-cn-southwest-2.oceanbase.cloud:3306/aptest',
    'username' = 'wktest', 
    'password' = '******',
    'table-name' = 'product_catalog',
    'sink.buffer-flush.max-rows' = '0',
    'sink.buffer-flush.interval' = '0');

INSERT INTO product_catalog SELECT * FROM mysql_product_catalog;
INSERT INTO orders_pay SELECT * FROM mysql_orders_pay;
INSERT INTO orders SELECT * FROM mysql_orders;
```

上記の Flink CDC 同期リンクを作成すると、データがリアルタイムで tptest データベースから aptest データベースに同期されます。SQL コンソールから該当のテーブルデータを確認できます。

コンピューティングレイヤーの構築

DBT プロジェクトの構築

元の TP テーブル構造はデータ分析および表示に直接使用するには適していないため、ある程度の変換が必要です。ここでは DBT プロジェクトを使用してデータ変換を行います。Demo のオリジナル例に基づいて DBT プロジェクトを構築し、モデルを定義します。詳細な手順は以下の通りです。

1. DBT のインストールおよびデプロイ後、dbt init my_project を実行して my_project という名前の DBT プロジェクトを作成し、ポップアップダイアログのプロンプトに従ってデータベース情報を入力します。ここで、ユーザーにはユーザー名のみを入力することに注意してください。作成後、先ほど入力した情報が /root/.dbt/profiles.yml ファイルに記録され、データベース接続に使用されます。

2. cd my_project を実行してプロジェクトディレクトリに入ります。

3. /my_project/models/example ディレクトリに SQL ファイルを記述して、データモデルを定義します。

   # models/example/dwd_orders.sql
   
   {{ config(
    materialized='view')
   }}
   
   select
       o.order_id as order_id,
       o.user_id as order_user_id,
       o.shop_id as order_shop_id,
       o.product_id as order_product_id,
       o.buy_fee as order_fee,
       o.create_time as order_create_time,
       o.update_time as order_update_time,
       o.state as order_state,
       c.catalog_name as order_product_catalog_name,
       p.pay_id as pay_id,
       p.pay_platform as pay_platform,
       p.create_time as pay_create_time
   from
       orders o
   left join product_catalog c on o.product_id = c.product_id
   left join orders_pay p on o.order_id = p.order_id
   

   # models/example/dwd_shops.sql
   {{
   config(materialized='table')
   }}
   
   select
       order_shop_id,
       str_to_date(date_format(pay_create_time, '%Y%m%d'), '%Y%m%d') as ds,
       sum(order_fee) as sum_fee
   from
       {{ ref('dwd_orders') }}
   where
       order_state=1
   GROUP BY
       order_shop_id,
   date_format(pay_create_time, '%Y%m%d')
   

   # models/example/dwd_shops_all.sql
   {{
   config(materialized='table')
   }}
   
   select
       str_to_date(date_format(pay_create_time, '%Y%m%d'), '%Y%m%d') as ds,
       sum(order_fee) as sum_fee
   from
       {{ ref('dwd_orders') }}
   where
       order_state=1
   GROUP BY
   date_format(pay_create_time, '%Y%m%d')
   

   # models/example/dwd_users.sql
   
   {{
   config(materialized='table')
   }}
   
   select
       order_user_id,
       str_to_date(concat(date_format(pay_create_time, '%Y%m'), '01'), '%Y%m%d') as ds,
       sum(order_fee) as sum_fee
   from
       {{ ref('dwd_orders') }}
   where
       order_state = 1
   group by
       order_user_id,
   date_format(pay_create_time, '%Y%m')
   

4. dbt compile を実行し、モデルをコンパイルします。つまり、SQL をデータベース固有のステートメントに変換します。

5. dbt run を実行して、すべてのモデルを実行します。

DAG スケジューリングタスクの作成

ここでは、基本的な DAG およびスケジューリングプラットフォームとして Airflow を採用します。初回のインストールおよびデプロイ後、airflow db init を実行して Airflow のデータベースを初期化する必要があります。Airflow をデプロイした後、airflow scheduler を実行して Scheduler を起動すると、DAG のリアルタイムスケジューリングが可能になります。airflow webserver を実行して Web サーバーを起動すると、Web インターフェース上でスケジューリングタスクの全体を確認できます。

Airflow の Web インターフェース（デフォルトポート 8080）にアクセスすることで、ワークフローでデータベースにアクセスできます。まず、Airflow の接続設定で OceanBase データベース接続を構成する必要があります。この手順は Airflow の Web UI で完了できます。データベースにアクセスする手順は以下の通りです。

1. Airflow Web UI を開きます。
2. Admin -> Connections にナビゲートします。
3. 「+」記号をクリックして新しい接続を追加します。
4. 以下のフィールドに記入します：
   • Connection Id: 接続ID。ここでは任意の識別子を指定できます。
   • Connection Type: 接続するデータベースタイプ。ここでは OceanBase の MySQL モードを例とするため、MySQL を入力する必要があります。
   • Host: OceanBase データベースのホストアドレス。ここでは接続文字列の -h の後のパラメータを入力します。
   • Schema: データベース名。ここでは接続する必要があるデータベースの名前を入力します。
   • Login: データベースユーザー名。ここでのフォーマットは username@tenantname です。
   • Password: データベースのパスワード。
   • Port: OceanBase ポート（デフォルトは 3306）

アクセスするデータベースを構成した後、バックグラウンドから Airflow がデプロイされているマシンにログインし、DAG ファイル（.py ファイル）を Airflow の dags ディレクトリに配置します。Airflow はこれらの DAG を自動的に検出して読み込みます。ここでは2つのコンピューティングノードを定義します。1つは DBT プロジェクトを実行してデータウェアハウスのコンピューティングタスクを完了させるため、もう1つはコンピューティングタスクの完了後にユーザーに通知メールを送信するためです。DAG の定義は以下の通りです。

# Copyright (c) 2023 OceanBase.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.email import EmailOperator
from airflow_dbt import DbtRunOperator

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "start_date": datetime(2024, 7, 12, 8, 35),
    "email": ["******@oceanbase.com"],
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG("warehouse_demo", default_args=default_args, schedule=timedelta(minutes=1))

t4 = EmailOperator(
    task_id="sending_email",
    to="******@oceanbase.com",
    subject="AirFlow Notice",
    html_content="<h1>Your Airflow Has been completed</h1>",
    dag=dag,
)

dbt_operator = DbtRunOperator(
    task_id="dbt_run", dir="xxx", dag=dag
)

dbt_operator >> t4

上記のファイルをデプロイすると、Airflow の Web インターフェースでスケジューリングタスクの全体を確認でき、必要に応じて該当タスクをスケジュールできます。

Grafana の統合

Prometheus データソースの追加

1. Grafana ホームページの左上角で、 アイコンをクリックして左側のサイドバーを展開し、Connections > Data sources を選択します。
2. Add data sources をクリックし、データソースとして Prometheus を選択します。
3. Prometheus データソース構成ページの Connection オプション配下で、Prometheus サーバーのアドレスを入力します。次にページ下部で Save & test をクリックします。

Grafana ダッシュボードの構成

1. アイコンをクリックして左側のサイドバーを展開し、**Dashboards** をクリックします。

2. + Create Dashboard をクリックし、作成ページの Import a dashboard カード内で Import Dashboard をクリックします。

3. ダウンロードした任意のダッシュボード設定ファイルをアップロードし、Load をクリックします。

4. Options ページでカスタムダッシュボード名を入力し、Import をクリックすると、監視ダッシュボードのデータを確認できます。または、必要に応じて単一の SQL データ情報を表示することもできます。ここでは店舗ごとの日別売上額のみを表示し、そのうちの1つの店舗の売上額とすべての店舗の売上額にのみ注目します。SQL は以下の通りです。

   SELECT
       ds AS "time",
       sum_fee AS "sum_fee"
   FROM dwd_shops
   WHERE
       order_shop_id = 35
   ORDER BY ds
   

実施

購買行動のシミュレーション

ユーザーの購買行動をシミュレーションするための簡単な Python スクリプト test.py を作成します。

# Copyright (c) 2023 OceanBase.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import dataclasses
from typing import Dict

import mysql.connector

PRODUCT_ID_2_FEE: Dict[int, float] = {
    1: 5399,
    2: 10099,
    3: 4599,
    4: 5499,
    5: 6499,
}

@dataclasses.dataclass
class Phone:
    product_id: int
    catalog_name: str
    buy_fee: float

def get_max_order_id(cur):
    cur.execute("select order_id from orders order by order_id desc limit 1")
    id = cur.fetchone()
    return next(iter(id))

def get_max_pay_id(cur):
    cur.execute("select pay_id from orders_pay order by pay_id desc limit 1")
    id = cur.fetchone()
    return next(iter(id))

def buy_phone(product_id: int, cursor, user_id=15, shop_id=35):
    cursor.execute("select product_id, catalog_name from product_catalog")
    tuples = cursor.fetchall()
    phones = [
        Phone(
            **{
                "product_id": p_id,
                "catalog_name": c_name,
                "buy_fee": PRODUCT_ID_2_FEE.get(p_id),
            }
        )
        for p_id, c_name in tuples
    ]
    target = next(filter(lambda p: p.product_id == product_id, phones))
    order_id = get_max_order_id(cursor) + 1
    sql = f"insert into `orders`(`order_id`, `user_id`, `shop_id`, `product_id`, `buy_fee`, `create_time`, `update_time`, `state`) values({order_id}, {user_id}, {shop_id}, {product_id}, {target.buy_fee}, now(), now(), 0)"
    cursor.execute(sql)
    pay_id = get_max_pay_id(cursor) + 1
    sql = f"insert into `orders_pay`(`pay_id`, `order_id`, `pay_platform`, `create_time`) values({pay_id}, {order_id}, 'Alipay', now())"
    cursor.execute(sql)
    sql = f"update orders set state=1 where order_id={order_id}"
    cursor.execute(sql)
    cursor.execute("commit")
    print(target)

if __name__ == "__main__":
    with mysql.connector.connect(
        **{
            "host": "******.huawei-cn-southwest-2.oceanbase.cloud",
            "port": "3306",
            "database": "tptest",
            "user": "wktest",
        }
    ) as conn:
        with conn.cursor() as cursor:
            buy_phone(5, cursor)

sh python.py コマンドを実行し、このスクリプトを実行します。

Airflow スケジューリング

1 分間待つと、Airflow は先ほど構成した DAG をスケジュールします。Web インターフェース上のステータスに基づいてノードの実行状態を判断でき、同時にタスクのログを確認できます。

データダッシュボードの確認

データダッシュボードを表示すると、最新のデータを確認できます。ここでは店舗の日別売上額のみを表示し、そのうちの1つの店舗の売上額とすべての店舗の売上額にのみ注目します。7月1日の売上額がすでに 32495 になっていることが確認できます。