#今回はEqualumのアップデート検証を行ってみます。
以前検証報告をさせて頂いた、kafka+SPARK環境を独自のノウハウと技術で統合し、先進のCDCエージェント技術を駆使してエンド・ツゥ・エンドでのExactly Onceを提供するEqualumなるソリューションに関して、最近大きなアップデートが有りましたので、取り急ぎ検証と結果共有をさせて頂きます。

#何が変わった？？
今回のバージョンでは、今までOracleのデータベース対応だけだったレプリケーション機能が、MySQL等の他のCDC対応データベースでの利用が可能になっています。また、以前のバージョンで対応していたOracle環境同様に、オリジナルの選択肢が単に増えただけではなく、レプリケーションの着地側データベースの選択肢も選べる様になっていますので、データシステム全体のコストバランスを考えた、より適切なシステム展開をサポートする事が可能です。

#では、レプリケーション機能を使ってみます。
今回の環境としては、

の構成を作ります。

##まずはデータの生成部分を作ります。
ここは、過去何度も使い回しているPython環境を”得意のバラック改造”してサクッと作ります。今回の環境も常設のPython3＋AnacondaをMBP上で使います。

###最初に空のテーブルを３種類作成します
＊　顧客系
＊　取引系
＊　決済系
ここも、エイヤー！でパラメータ定義をしていますので、気になられる方は適宜書き換えてください（汗）。今回は、共通キー的なカラムとして、全体を透過的に共通化しているオーダーID（Order_ID）を入れています。

# coding: utf-8
#
# MySQLにオリジナル側のCDCテーブルを作成
#
#　Python3版
#

#  初期設定
import sys
stdout = sys.stdout
sys.stdout = stdout

import pymysql.cursors

# テーブル定義　
# ヘッダー系情報
DC0 = "id INT AUTO_INCREMENT, ts TIMESTAMP(6), PRIMARY KEY(id, ts), Order_ID VARCHAR(15), "
# 顧客情報
DC1 = "User VARCHAR(20), Zip VARCHAR(10), Prefecture VARCHAR(10), Address VARCHAR(60), "
DC2 = "Tel VARCHAR(15), Email VARCHAR(40), Point INT, Area VARCHAR(10)"
# 取引系情報
DC3 = "Category VARCHAR(20), Product VARCHAR(20), Price INT, Units INT, Logistics VARCHAR(20) "
# 決済系情報
DC4 = "User VARCHAR(20), Card VARCHAR(40), Number VARCHAR(30), Price INT, Units INT, Payment INT, Tax INT "
# テーブル名
Table_Name = ["Rep_Demo_Usr_Table","Rep_Demo_Biz_Table","Rep_Demo_Pay_Table"]
# 作成するテーブル数
Generate_Table = 3

try:
   
    print("オリジナル側CDCテーブル作成処理を開始")
    
    # テーブル定義の初期化
    Table_Create = []
    Create_SQL = "CREATE TABLE IF NOT EXISTS " + Table_Name[0] + " ("+DC0+DC1+DC2+")"  
    Table_Create.append(Create_SQL)    
    Create_SQL = "CREATE TABLE IF NOT EXISTS " + Table_Name[1]+ " ("+DC0+DC3+")"
    Table_Create.append(Create_SQL)
    Create_SQL = "CREATE TABLE IF NOT EXISTS " + Table_Name[2] + " ("+DC0+DC4+")" 
    Table_Create.append(Create_SQL)
   
    # MySQLとの接続
    db = pymysql.connect(host = 'xxx.xxx.xxx.xxx',
                         port=3306,                
                         user='xxxxxxxx',
                         password='zzzzzzzz',
                         db='xxxxxxxx',        
                         charset='utf8mb4',
                         cursorclass=pymysql.cursors.DictCursor)
                                      
    with db.cursor() as cursor:
        
        Counter = 0
        
        while Counter < Generate_Table:
       
            # テーブルの初期化
            cursor.execute("DROP TABLE IF EXISTS " + Table_Name[Counter])
            db.commit()
            
            #　新規テーブルを作成
            cursor.execute(Table_Create[Counter])    
            db.commit()
            
            Counter = Counter + 1
        
except KeyboardInterrupt:
   
    print('!!!!! 割り込み発生 !!!!!')
           
finally:
   
    # データベースコネクションを閉じる
    db.close()
   
    print("オリジナル側CDCテーブル作成処理が終了")

###次にデータを連続挿入する仕組みを作ります。
顧客系に関しては、実際の取引とは関係ない「単純な顧客登録のみ」が時々入ってくる様にしました。またこの割合は

if str(fakegen.boolean(40)) == "True":

の部分で４０の値を変える事で可能です（詳しくは、Pythonのマニュアルにて・・（汗））
また、リアル感を追加する為に、ランダムな暇つぶし処理も入っていますので、これも適宜書き換えて頂いて結構です。
（因みに、Equalum的には一番ドSな待ち時間無しの設定でも普通に処理します・・・・・）

# coding: utf-8
#
# MYSQLの各CDCテーブルをランダムに更新する
#
#　Python３版
#

#  初期設定
import sys
stdout = sys.stdout
sys.stdout = stdout

import time
import pymysql.cursors
import re

# デモで使うメタデータ定義
Category_Name = ["酒類","家電","書籍","DVD/CD","雑貨"]

Product_Name0 = ["日本酒","バーボン","ビール","芋焼酎","赤ワイン","白ワイン","スコッチ","ブランデー","泡盛","テキーラ"]
Product_Price0 = [1980, 2500, 490, 2000, 3000, 2500, 3500, 5000, 1980, 2000]    

Product_Name1 = ["テレビ","洗濯機","ラジオ","ステレオ","電子レンジ","パソコン","電池","エアコン","乾燥機","掃除機"]
Product_Price1 = [49800, 39800, 2980, 88000, 29800, 64800, 198, 64800, 35800, 24800]    

Product_Name2 = ["週刊誌","歴史","写真集","漫画","参考書","フィクション","経済","自己啓発","月刊誌","新刊"]
Product_Price2 = [280, 1500, 2500, 570, 1480, 1400, 1800, 1540, 980, 1980]    

Product_Name3 = ["洋楽","演歌","Jポップ","洋画","アイドル","クラッシック","邦画","連続ドラマ","企画","アニメ"]
Product_Price3 = [1980, 2200, 2500, 3500, 2980, 1980, 3800, 2690, 1980, 2400]    

Product_Name4 = ["洗剤","電球","贈答品","医薬部外品","ペットフード","乾電池","文房具","男性用品","女性用品","季節用品"]
Product_Price4 = [498, 198, 1980, 398, 980, 248, 398, 2980, 3580, 1980]

Table_Name1 = "Rep_Demo_Usr_Table" # 顧客管理系テーブル
Table_Name2 = "Rep_Demo_Biz_Table" # 販売・物流系テーブル
Table_Name3 = "Rep_Demo_Pay_Table" # 決済系テーブル

#　地域名ルックアップ情報（キーは都道府県名）
Area_Data={'北海道':'北海道','青森県':'東北','岩手県':'東北','宮城県':'東北','秋田県':'東北','山形県':'東北','福島県':'東北',
           '茨城県':'関東','栃木県':'関東','群馬県':'関東','埼玉県':'関東','千葉県':'関東','東京都':'関東','神奈川県':'関東',
           '新潟県':'中部','富山県':'中部','石川県':'中部','福井県':'中部','山梨県':'中部','長野県':'中部','岐阜県':'中部','静岡県':'中部','愛知県':'中部',
           '三重県':'近畿','滋賀県':'近畿','京都府':'近畿','大阪府':'近畿','兵庫県':'近畿','奈良県':'近畿','和歌山県':'近畿',
           '鳥取県':'中国','島根県':'中国','岡山県':'中国','広島県':'中国','山口県':'中国',
           '徳島県':'四国','香川県':'四国','愛媛県':'四国','高知県':'四国',
           '福岡県':'九州・沖縄','佐賀県':'九州・沖縄','長崎県':'九州・沖縄','熊本県':'九州・沖縄','大分県':'九州・沖縄','宮崎県':'九州・沖縄','鹿児島県':'九州・沖縄','沖縄県':'九州・沖縄'}

# 物流センタールックアップ情報（キーは地域名）
Logi_Data = {'北海道':'道央物流センター','東北':'東北物流センター','関東':'関東中央物流センター',
             '中部':'甲州物流センター','近畿':'伊丹物流センター','中国':'広島臨港物流センター','四国':'讃岐物流センター','九州・沖縄':'平戸物流センター'}

# 購入ポイント情報（カテゴリ名の順番に設定
Point_Data = [0.02, 0.1, 0.03, 0.02, 0.05]

# 消費税率の設定
Tax_Data = 0.1

# 生成するデータの総数
Loop_Count = 10

# タイミング調整フラグ　（０：待ち時間無し　１：1秒おき　２：ランダム）
Wait_Flag = 2

# 一定間隔の場合（システム時間で秒単位）
Sleep_Wait = 1
    
# ランダム間隔の場合（実態に合わせて調整）
#Base_Count = 1600000
Base_Count = 800000

# SQL情報をコンソールに表示するか否かの設定（１：表示する）
Display_SQL = 0

# 書き込み用のデータカラム
DL0 = "Order_ID, "
DL1 = "User, Zip, Prefecture, Address, Tel, Email, Point, Area"
DL2 = "Category, Product, Price, Units, Logistics"
DL3 = "User, Card, Number, Price, Units, Payment, Tax"

def Add2Pre(Address_Data):
    
    # 都道府県情報の抽出
    pattern = u"東京都|北海道|(?:京都|大阪)府|.{2,3}県"
    m = re.match(pattern , Address_Data)
    if m:
        Prefecture_Name = m.group()
        
    return(Prefecture_Name)

try:
   
    print("テーブルへ連続挿入を開始")

    # Fakerの初期化
    from faker import Faker
    fakegen = Faker('ja_JP')    
    Faker.seed(fakegen.random_digit())
   
    # その他の変数
    Counter = 0
    Work_Count = 1
       
    # MySQLとの接続
    db = pymysql.connect(host = 'xxx.xxx.xxx.xxx',
                         port=3306,                
                         user='xxxxxxxx',
                         password='zzzzzzzz',
                         db='xxxxxxxx',        
                         charset='utf8mb4',
                         cursorclass=pymysql.cursors.DictCursor)
   
    with db.cursor() as cursor:
   
        # 検証データの生成
        while Counter < Loop_Count:
           
            # ランダムに書き込む商材の種類を選択
            Category_ID = fakegen.random_digit()
            if Category_ID > 4: Category_ID = Category_ID - 5
                
            # カテゴリ名の設定  
            Category = Category_Name[Category_ID]
            # 商品IDの設定
            Product_ID = fakegen.random_digit()
            
            # オーダーIDの情報作成
            S_Counter = str(Counter)
            O_Counter = S_Counter.zfill(8)
           
            if Category_ID == 0:
                Product = Product_Name0[Product_ID]
                Price = Product_Price0[Product_ID]
                Units = fakegen.random_digit() + 1
                Point = Price * Units * Point_Data[Category_ID]
                O_ID = "C00" + O_Counter
               
            elif Category_ID == 1:
                Product = Product_Name1[Product_ID]
                Price = Product_Price1[Product_ID]
                Units = 1
                Point = Price * Units * Point_Data[Category_ID]
                O_ID = "C01" + O_Counter
               
            elif Category_ID == 2:
                Product = Product_Name2[Product_ID]
                Price = Product_Price2[Product_ID]
                Units = fakegen.random_digit() + 1
                if Units >3: Units = 3  
                Point = Price * Units * Point_Data[Category_ID]
                O_ID = "C02" + O_Counter
                   
            elif Category_ID == 3:
                Product = Product_Name3[Product_ID]
                Price = Product_Price3[Product_ID]
                Units = fakegen.random_digit() + 1
                if Units >2: Units = 2
                Point = Price * Units * Point_Data[Category_ID]
                O_ID = "C03" + O_Counter
                   
            else:
                Product = Product_Name4[Product_ID]
                Price = Product_Price4[Product_ID]
                Units = fakegen.random_digit() + 1
                if Units >4: Units = 4
                Point = Price * Units * Point_Data[Category_ID]
                O_ID = "C04" + O_Counter
            
            # 顧客名
            User = fakegen.name()
            
            # 支払い情報
            if str(fakegen.pybool()) == "True":
                Card = "現金"
            else:
                Card = fakegen.credit_card_provider()
   
            Number = fakegen.credit_card_number()              
            if Card == "現金":    Number = "N/A"
            
            # 支払い総額と消費税
            Payment = Price * Units
            Tax = Payment * 0.1

            # 顧客情報の設定
            Zip = fakegen.zipcode()
            Address = fakegen.address()            
            Prefecture = Add2Pre(Address)           
            Tel = fakegen.phone_number()
            Email = fakegen.ascii_email()
            
            # 地域名と物流センター名を取得           
            Area = Area_Data.get(Prefecture)
            Logistics = Logi_Data.get(Area)
        
            # 購入処理系は随時全て更新し、顧客管理のみ時々顧客登録のみを実施する            
            if str(fakegen.boolean(40)) == "True":
                
                Point = 0 # 登録だけなのでポイントは無し
                
                # 顧客登録のみのコードを設定
                O_ID = "W99" + O_Counter
                
                DV0 = O_ID + "','"                
                DV1 = User + "','" + Zip + "','" + Prefecture + "','" + Address + "','" + Tel + "','" + str(Email) + "','" + str(Point) + "','" + Area
                SQL_Data1 = "INSERT INTO " + Table_Name1 + " (" + DL0 + DL1+ ") VALUES ('"+DV0 + DV1 + "')"
                
                # 顧客登録のみを処理
                cursor.execute(SQL_Data1)
                db.commit()
                
                #  コンソールに生成データを表示
                if Display_SQL == 1:
                    print (SQL_Data1) 
                    
            else: # 通常の処理
                
                # 挿入データ用SQLの準備
                DV0 = O_ID + "','"
                DV1 = User + "','" + Zip + "','" + Prefecture + "','" + Address + "','" + Tel + "','" + str(Email) + "','" + str(Point) + "','" + Area
                DV2 = Category + "','" + Product + "','" + str(Price) + "','" + str(Units) + "','" + Logistics
                DV3 = User + "','" + Card + "','" + Number + "','" + str(Price) + "','" + str(Units) + "','" + str(Payment) + "','" + str(Tax)
                
                # SQLの作成
                SQL_Data1 = "INSERT INTO " + Table_Name1 + " (" + DL0 + DL1+ ") VALUES ('"+DV0 + DV1 + "')"
                SQL_Data2 = "INSERT INTO " + Table_Name2 + " (" + DL0 + DL2 + ") VALUES ('"+DV0 + DV2 + "')"
                SQL_Data3 = "INSERT INTO " + Table_Name3 + " (" + DL0 + DL3 + ") VALUES ('"+DV0 + DV3 + "')"
                
                # MySQLへ書き込む          
                cursor.execute(SQL_Data1)
                db.commit()
            
                cursor.execute(SQL_Data2)
                db.commit()
            
                cursor.execute(SQL_Data3)
                db.commit()
                                   
                #  コンソールに生成データを表示
                if Display_SQL == 1:
                    print (SQL_Data1)
                    print (SQL_Data2)
                    print (SQL_Data3)
                
           
            # 生成間隔の調整
            if Wait_Flag == 1:
                time.sleep(Sleep_Wait)
            elif Wait_Flag == 2:
                Wait_Loop = Base_Count * fakegen.random_digit() + 1
                for i in range(Wait_Loop): Work_Count = Work_Count + i
       
            # ループカウンタの更新
            Counter=Counter+1
            
            # データの作成状況を表示
            if (Counter % 10) == 0:     print("途中経過：　" + str(Counter) + " 個目のデータ作成を終了")

except KeyboardInterrupt:
   
    print('!!!!! 割り込み発生 !!!!!')

finally:
   
    # データベースコネクションを閉じる
    db.close()
   
    print("生成したデータの総数 : " + str(Counter))    
    print("テーブルへ連続挿入を終了")

##取り敢えずのテストを実施・・・
では、取り急ぎのテストを実施してみます。今回は仮想環境上にCDC設定済みのMySQLを立ち上げて、その上にレプリケーション検証用のデータベースを作成し、その中に3個のテーブルを作成します。

無事にテーブルが出来ましたので、今度は連続挿入の処理を走らせてみます。ここではランダム感覚で１０個のデータを生成・挿入します。

「販売・物流系テーブル」

「決済系テーブル」

「顧客系テーブル」

上手くデータが分類されて、３種類のテーブルが出来ている事が確認出来ました。

#いよいよEqualumの登場！
この画面は、以前検証報告させて頂いたEqualumと「殆ど同じ」になりますが、今回のレプリケーション機能の対応拡大に伴い、２箇所大きな追加・変更が有ります。

まずは右上の部分

左側のメニューバー部分

になります。
##レプリケーションの設定をしてみる。。。
まずは、左側のメニューバーからレプリケーションを選択します。

+ADDボタンを選択します。

あとは、上から順番に必要項目を選択・設定していけば終了です。
（１）グループ名を設定してソースとターゲット、その他の項目を設定
（２）Add Tablesの下に**+ボタンが有るので選択します。

（３）必要項目を選択・入力します。今回はRepxxxxxという形でテーブル名を統一していますので、Rep%と設定しました。この部分は、その他の省略形式もサポートしていますので、関連するテーブルを一気に選択・設定する際に便利な機能になります。
（４）Generate Listボタンを選択すると、先程の条件に適合するテーブルが自動的に選択されますので、問題がなければValidate Tableを選択して接続確認します。

無事に整合性が取れていると判断された場合には、緑色のチェックマークが付きますのので、右下のCreate**ボタンを押して登録します。（万が一この様な警告が出た場合は、ターゲット側に同じ名前のテーブルが無いかどうかを確認してください。）

##では検証開始！
無事に設定・登録が完了すると、管理画面にEqualumが自動生成したFLOWとして３個のストリーミング・レプリケーションが確認出来ます。

まずは、DBeaverを使ってスタートの状況を確認します。

空っぽのテーブルが、先程のPythonスクリプトで作成され、この段階ではターゲットのSingleStore側には何も変化は有りませんが、先程の連続挿入スクリプトを走らせると、ターゲット側にテーブルが作られてデータがストリーミングでレプリケーションされます。

念の為に、DBeaverで確認してみます。
ソース側のテーブル

ターゲット側のテーブル

無事にレプリケーションが行われている様です。

#今回のまとめ
今回は、Equalumのレプリケーション機能についての簡単な動作検証を行いました。以前の記事を読まれた方で”勘の良い方”はピン！ときたかもしれませんが、Equalumの場合途中のFLOW処理を入れなければ、基本的にはレプリケーションと同じ”様な”動作になります。もちろん、上流側のソースDBがCDCストリーミングをサポートしていれば、”ほぼ同じ”様な動作を手組みで作る事も可能なのですが、現実的なレプリケーション環境を考えると、命名規則に則ったテーブル名設定や、それらのリレーションを保持する為にテーブルをグループ化して管理しなければならない・・といった要求仕様が出てきます。
今回Oracle環境以外に横展開を始めたEqualumのレプリケーション環境では、出来るだけ利用者の負担を減らす方向で、レプリケーションに合わせた機能追加などを行い、通常メニューとして利用出来る様になりました。
また、冒頭にも書かせて頂いた通り、同じデータベース同志でなければレプリケーション出来ない！という仕様でも有りませんので、用途やコスト見合いで柔軟に構成を選択する事が可能です。

#さて次回は・・・・
次回は、今回検証出来なかった、レプリカ側に対してPythonを使った簡単な即時可視化の仕掛けを入れて、利活用を意識した検証を行いたいと思います。もちろん、シンプルなBCP対応！という事であれば、今回の検証で十分ご検討頂けるかと思いますが、逆に「取り敢えず現場から全部持ってきて、それから考える！（ETL的処理は会議室で考える！系）というアプローチも否定できませんので、（この辺は、Equalumが稼働するサーバ・クラスターの構成にも関係しますが・・・）興味本意半分以上ではありますが、可視化ツールの作成から作業を行ってみたいと思います。

#謝辞
本検証は、Equalum社の公式バージョン（V2.２３)を利用して実施しています。
この貴重な機会を提供して頂いたEqualum社に対して感謝の意を表すると共に、本内容とEqualum社の公式ホームページで公開されている内容等が異なる場合は、Equalum社の情報が優先する事をご了解ください。