OB Cloudに基づくオフラインデータウェアハウスと可視化ダッシュボードの構築

背景

社会のデジタル化が加速する中、企業におけるデータ分析への要求はますます高まっています。データ処理は主に大量の履歴データを扱うオフラインシナリオに集中しており、データウェアハウスの構築によって解決できます。OB Cloudはデータウェアハウスを構築することで、オフラインデータの分析課題を解決します。スケジュールタスクを通じて、生データ層（ODS）から明細データ層（DWD）、最終的にアプリケーションデータ層（ADS）へと至るデータ階層を構築し、OB Cloudベースのデータウェアハウスを作成します。同時に、関連するエコシステムツールを利用して可視化ダッシュボードを構築できます。

前提条件

• ご自身は現在のインスタンスに対するインスタンス管理者およびデータ読取り権限を保有しています。権限がない場合は、組織管理者に連絡し追加してもらえます。
• 利用可能な分析型（MySQL）クラスタインスタンスをお持ちです。テナントの作成については、テナントの作成をご参照ください。
• データベースカーネルのバージョンが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 アカウントを選択し、パスワードを入力してログインします。OK をクリックします。

   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プロジェクトを作成します。表示されるダイアログボックスで、プロンプトに従ってデータベース情報を入力します。ここで注意が必要なのは、userにはユーザー名のみを入力することです。作成後、先ほど入力した情報がファイル /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スケジュールタスクの作成

ここでは、Airflowを基本的なDAGおよびスケジューリングプラットフォームとして採用します。初回インストール・デプロイ後、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. アップロードまたはダウンロードしたいダッシュボードの設定ファイルを選択し、ロード をクリックします。

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日の売上高は既に32,495になっていることがわかります。