【TROCCO】Snowflakeで加工したデータをDatabricksのDashBoardsで表示してみる！

はじめに

【Databricks】Dashboardsでピボットテーブルを触ってみるでピボットテーブルを使ってみるといい感じだったので、TROCCO&COMETAを触ってみた！で作成したデータをDatabricksに転送してDashBoardsで表示してみました！

DatabricksのDashBoardsで作成したピボットテーブル

各SaaSの役割として

• データの保存：Snowflake
• データの表示：Databricks
• 転送・加工・トリガー(ワークフローオーケストレーションツール)：TROCCO

というような感じで作成しています。

TROCCOのワークフロー

今回は上記の作業中につまづきポイントがあったため備忘録として残しておきます。

事前準備

• Snowflake
  • こちらを参考
• Databricks
  • 下記を事前に作成する必要があります
    • アカウント
    • Compute
    • SQL warehouses
    • カタログ(DBのようなもの)
    • スキーマ
• TROCCO：アカウントが作成済みであること

TROCCOの接続設定

ここではDatabricksとTROCCOの接続設定のみを記載します。

SnowflakeとTROCCOの設定に関してはこちらを見てもらえると嬉しいです。

Databricks

接続情報→DWH→Databricksを選択

下記の項目を設定していきます。

• Server Hostname
  • DataBricksのCompute→<Compute名>→Advanced options→JDBC/ODBCから取得可能
• HTTP Path
  • 同上

• Personal Access Token
  • Userレベルでアクセストークンを作成する
  • DataBricksのユーザアイコン→settings→User→Developer→Access token→Generate new tokenから作成可能

Lifetimeは入力無しで Expiration Never となります。
認証の期限が必要であれば入力したほうが良いですね。

上記3点を設定して接続を確認をクリックし、通信が通っていればOK。

TROCCOの転送設定

Snowflake→Databricks

転送設定→新規転送設定作成→転送元：Snowflake→転送先：databricks→この内容で作成を選択する。

共通で設定する項目

• 名前：任意
  • ワークフローで検索しやすいプリフィックスがあると楽です。 (Snowflake→Databricksのようなもの)

Snowflakeで設定する項目

• Snowflake接続情報：こちらで作成した接続情報を使います
• データベース：<任意のDB>
• スキーマ：MART
• クエリ：select * from mart_jaffle_shop_pivot_table_trocco

Databricksで設定する項目

• Databricks接続情報：先ほど作成したDatabricksの接続設定
• カタログ：<任意のカタログ名(DBのようなもの)>
  • 事前に作成必須
• スキーマ：mart
  • 事前に作成必須
  • 名前は任意ですがSnowflakeと合わせると楽です
• テーブル名：mart_jaffle_shop_pivot_table_trocco
  • 任意ですがSnowflakeと合わせると楽です
  • テーブルが存在しない場合、TROCCOが自動で作成してくれます。
• 転送モード：全件洗い替え (REPLACE)
  • 連携するデータが少ないならSnowflakeと常に同期するように全件洗替にすると設計は楽です

データのプレビュー・型の設定

下記の設定でデータを転送します。

• カラムの型：String
• カラム名：ローマ字表記

Databricksでもquoteを使えば日本語名カラムを取り扱えるのですが、
TROCCOだとちょっとうまくいかなかったのでローマ字に置き換えてます。

購入履歴管理IDや `購入履歴管理ID`の入力を試しましたが同じエラーで怒られるため、
ここは大人しくローマ字で行きましょう！

エラー内容

[INVALID_IDENTIFIER] The unquoted identifier 購入履歴管理ID is invalid and must be back quoted as: `購入履歴管理ID`.
Unquoted identifiers can only contain ASCII letters ('a' - 'z', 'A' - 'Z'), digits ('0' - '9'), and underbar ('_').
Unquoted identifiers must also not start with a digit.

ただ、ローマ字に変更して成功処理を見ているとカラム名にquoteがしっかりと適用されているのでちょっと不思議な感じですね〜

 [INFO] (0001:transaction): SQL: CREATE TABLE `mart`.`mart_jaffle_shop_pivot_table_trocco_000001935916a0fe_embulk` (`KOUNYUU_RIREKI_KANRI_ID` STRING, `SKU` STRING, `KOKYAKU_KANRI_ID` STRING, `ZEINUKI_URIAGE` STRING, `ZEIKIN_NO_URIAGE` STRING, `ZEIKOMI_URIAGE` STRING, `SHOUHIN_NO_TORIHIKI_HIZUKE` STRING, `YOUBI_CODE` STRING, `YOUBI` STRING, `SHUKUJITSU_MEI` STRING, `DO_NICHI_SHUKU` STRING, `GW` STRING, `GW_CHUUKAN_MO_YASUMI` STRING, `NENMATSU_NENSHI` STRING, `RENKYUU` STRING, `KANSOKUSHO_MEI` STRING, `HEIKIN_KION` STRING, `HEIKIN_SHITSUDO` STRING, `TENKI_GAIKYOU_HIRU` STRING, `TENKI_GAIKYOU_YORU` STRING, `SEIBETSU` STRING, `NENDAI` STRING, `E_ME_RU` STRING, `SHOUHIN_MEI` STRING)

動作確認

転送設定→Snowflake→Databricks mart_jaffle_shop_pivot_table→実行→実行を選択して、転送できていればOK！

ちなみにDatabricksは初回の立ち上げに時間がかかるため、
2回目以降は爆速で転送ができます。

SnowflakeではWHが高速で立ち上がるので意識はしていませんでしたが、
DatabricksのComputeは立ち上がりが遅めなのでTROCCOで余分な時間課金が発生する可能性があります。
(転送の総時間は、転送時間+5~10分位 を見ておくと良さそうです。)

ただ、DatabricksのComputeを立ち上げっぱなしにするとDBU+クラウドインフラの起動時間と恐ろしい課金が発生するためComputeの常時立ち上げはおすすめしません。

なので、本番環境でany→Databricksの転送スケジュールを組む際は、
キュー待ちが発生しない程度に同時間帯・同Computeを使って転送しちゃう作戦を取ると思います。

ワークフローへの組み込み

既存ワークフローへの転送設定追加

こちらで作成したワークフローに今回の転送設定を組み込みます。

ワークフロー→mart_jaffle_shop_pivot_table→フロー編集から転送設定のSnowflake→Databricks mart_jaffle_shop_pivot_tableを追加します。

あとは保存→保存を適用をクリックして作成完了です。

動作確認

ワークフロー→mart_jaffle_shop_pivot_table→実行→実行をクリックしてワークフローを動作させます。

全てのジョブが成功していればOKです！

DatabricksのDashBoardsでデータ表示

DashBoards上でのデータ加工

現状はDataBricksに下記のような形式でデータが存在しています。

• カラムの型：String
• カラム名：ローマ字

これをSnowflakeの型と近い形に直していきます。

• カラムの型：数値・Boolean・Date等にcast
• カラム名：日本語

DashBoards→<DashBorad名>→Data→Create from SQLからDatasetを作成します。

TROCCOのデータマート定義にDatabricksが来たら下記のSQLもTROCCO側に寄せられると思います。

WITH 

mart_jaffle_shop_pivot_table_trocco AS (
  SELECT　*　FROM　<カタログ名>.mart.mart_jaffle_shop_pivot_table_trocco
),
rename_col AS (
  SELECT
    KOUNYUU_RIREKI_KANRI_ID AS `購入履歴管理ID`,
    SKU AS `SKU`,
    KOKYAKU_KANRI_ID AS `顧客管理ID`,
    ZEINUKI_URIAGE AS `税抜き売上`,
    ZEIKIN_NO_URIAGE AS `税金の売上`,
    ZEIKOMI_URIAGE AS `税込み売上`,
    SHOUHIN_NO_TORIHIKI_HIZUKE AS `商品の取引日付`,
    YOUBI_CODE AS `曜日コード`,
    YOUBI AS `曜日`,
    SHUKUJITSU_MEI AS `祝日名`,
    DO_NICHI_SHUKU AS `土日祝`,
    GW AS `GW`,
    GW_CHUUKAN_MO_YASUMI AS `GW_中間も休み`,
    NENMATSU_NENSHI AS `年末年始`,
    RENKYUU AS `連休`,
    KANSOKUSHO_MEI AS `観測所名`,
    HEIKIN_KION AS `平均気温`,
    HEIKIN_SHITSUDO AS `平均湿度`,
    TENKI_GAIKYOU_HIRU AS `天気概況_昼`,
    TENKI_GAIKYOU_YORU AS `天気概況_夜`,
    SEIBETSU AS `性別`,
    NENDAI AS `年代`,
    E_ME_RU AS `Eメール`,
    SHOUHIN_MEI AS `商品名`
  FROM
    mart_jaffle_shop_pivot_table_trocco
),
cast_type AS (
  SELECT
    CAST(`購入履歴管理ID` AS STRING) AS `購入履歴管理ID`,
    CAST(`SKU` AS STRING) AS `SKU`,
    CAST(`顧客管理ID` AS STRING) AS `顧客管理ID`,
    CAST(`税抜き売上` AS DECIMAL(38, 0)) AS `税抜き売上`,
    CAST(`税金の売上` AS DECIMAL(38, 0)) AS `税金の売上`,
    CAST(`税込み売上` AS DECIMAL(38, 0)) AS `税込み売上`,
    CAST(`商品の取引日付` AS DATE) AS `商品の取引日付`,
    CAST(`曜日コード` AS STRING) AS `曜日コード`,
    CAST(`曜日` AS STRING) AS `曜日`,
    CAST(`祝日名` AS STRING) AS `祝日名`,
    CAST(`土日祝` AS BOOLEAN) AS `土日祝`,
    CAST(`GW` AS BOOLEAN) AS `GW`,
    CAST(`GW_中間も休み` AS BOOLEAN) AS `GW_中間も休み`,
    CAST(`年末年始` AS BOOLEAN) AS `年末年始`,
    CAST(`連休` AS BOOLEAN) AS `連休`,
    CAST(`観測所名` AS STRING) AS `観測所名`,
    CAST(`平均気温` AS DECIMAL(38, 0)) AS `平均気温`,
    CAST(`平均湿度` AS DECIMAL(38, 0)) AS `平均湿度`,
    CAST(`天気概況_昼` AS STRING) AS `天気概況_昼`,
    CAST(`天気概況_夜` AS STRING) AS `天気概況_夜`,
    CAST(`性別` AS STRING) AS `性別`,
    CAST(`年代` AS STRING) AS `年代`,
    CAST(`Eメール` AS STRING) AS `Eメール`,
    CAST(`商品名` AS STRING) AS `商品名`
  FROM
    rename_col
),
final AS (
  SELECT
    `購入履歴管理ID`,
    `SKU`,
    `顧客管理ID`,
    `税抜き売上`,
    `税金の売上`,
    `税込み売上`,
    `商品の取引日付`,
    `曜日コード`,
    `曜日`,
    `祝日名`,
    `土日祝`,
    `GW`,
    `GW_中間も休み`,
    `年末年始`,
    `連休`,
    `観測所名`,
    `平均気温`,
    `平均湿度`,
    `天気概況_昼`,
    `天気概況_夜`,
    `性別`,
    `年代`,
    `Eメール`,
    `商品名`
  FROM
    cast_type
)
SELECT　*　FROM　final

DashBoardsのCanvasで表示

作成したDatasetからレコード数・ピボットテーブルのグラフを作成します。
各カラムの型はDataset内のSQLで加工しているので今回は載せてません。

1.7Mほどのデータなので処理は早いですが、「週や日ごとに集計データが見たい」のような処理をすると表示までに5秒ほど固まりますね。。

分ける要素の数にもよりますが細かい粒度で分ける際は表示速度でも時間がかかっているんだな〜と思ってあげると優しくなれます。

これで、SnowflakeからDatabricksへデータを転送してDashBoardsのCanvasで表示が完了しました！

さいごに

今回はSnowflakeからDatabricksへデータを転送してDashBoardsのCanvasでデータを表示しました。

TROCCOとSnowflakeの連携は多めに使っているためエラーの勘所があるのですが、
TROCCOとDatabricksの連携は初めてだったので様々なつまづきポイントに気づけて良かったです。

あとこの頃、「TROCCOのワークフローが小規模開発のオーケストレーションツールとしてありなのでは！？」と勝手に思っているので試していないサービス同士の連携を試すのが楽しくなってきました笑

また、IaC対応も少しずつ進んでいるそうなので試す楽しみが増えて嬉しいところですね！

開発の規模をスケールしようとするとIaC(一括変更やCI等)が欲しくなるので検証したくなる・・・

横道はこのぐらいにして、今回の記事を見て是非試したいかたはTROCCOを使ってみてください。
以上hamaでした〜