理想的なデータ ウェアハウスはどのような原則に基づいて構築されていますか?

定型的なコードがない中で、ビジネス価値と分析に焦点を当てます。コードベースとしての DWH の管理: バージョン管理、レビュー、自動テスト、CI。モジュール式、拡張可能、オープンソース、コミュニティ。ユーザーフレンドリーなドキュメントと依存関係の視覚化 (データリネージュ)。

これらすべてと、ビッグデータと分析エコシステムにおける DBT の役割について詳しくは、cat へようこそ。

もしもし

アルテミー・コズィルから連絡があります。 5 年以上、私はデータ ウェアハウス、ETL/ELT の構築、データ分析と視覚化に取り組んできました。私は現在働いています ウィリー, 私はOTUSでコースを教えています Data Engineerそして今日は、開始を予期して書いた記事を共有したいと思います コースへの新規登録.

概要

DBT フレームワークは、ELT (抽出 - 変換 - ロード) の頭字語の T がすべてです。

BigQuery、Redshift、Snowflake などの生産性とスケーラブルな分析データベースの出現により、データ ウェアハウスの外で変換を行う意味がなくなりました。 

DBT はソースからデータをダウンロードしませんが、既にストレージ (内部ストレージまたは外部ストレージ) にロードされているデータを操作する優れた機会を提供します。

DBT の主な目的は、コードを取得して SQL にコンパイルし、リポジトリ内で正しい順序でコマンドを実行することです。

DBT プロジェクトの構造

プロジェクトは、次の 2 種類のみのディレクトリとファイルで構成されます。

• モデル (.sql) - SELECT クエリによって表現される変換の単位
• 構成ファイル (.yml) - パラメーター、設定、テスト、ドキュメント

基本的なレベルでは、作業は次のように構成されています。

• ユーザーは任意の便利な IDE でモデル コードを準備します。
• CLI を使用してモデルが起動され、DBT がモデル コードを SQL にコンパイルします。
• コンパイルされた SQL コードはストレージ内で指定された順序で実行されます (グラフ)

CLI から実行すると次のようになります。

すべてはSELECTです

これは、データ構築ツール フレームワークの重要な機能です。言い換えれば、DBT は、クエリをストアに具体化することに関連するすべてのコードを抽象化します (コマンド CREATE、INSERT、UPDATE、DELETE ALTER、GRANT などからのバリエーション)。

どのモデルでも、結果のデータ セットを定義する 1 つの SELECT クエリを作成する必要があります。

この場合、変換ロジックはマルチレベルにすることができ、他のいくつかのモデルからのデータを統合できます。注文ショーケース (f_orders) を構築するモデルの例:

{% set payment_methods = ['credit_card', 'coupon', 'bank_transfer', 'gift_card'] %}
 
with orders as (
 
   select * from {{ ref('stg_orders') }}
 
),
 
order_payments as (
 
   select * from {{ ref('order_payments') }}
 
),
 
final as (
 
   select
       orders.order_id,
       orders.customer_id,
       orders.order_date,
       orders.status,
       {% for payment_method in payment_methods -%}
       order_payments.{{payment_method}}_amount,
       {% endfor -%}
       order_payments.total_amount as amount
   from orders
       left join order_payments using (order_id)
 
)
 
select * from final

ここではどんな興味深いものを見ることができるでしょうか?

最初: CTE (共通テーブル式) を使用 - 多くの変換とビジネス ロジックを含むコードを整理して理解するために

2 番目: モデル コードは SQL と言語の混合物である ジンジャ (テンプレート言語)。

この例ではループを使用しています の 式で指定された各支払い方法の金額を生成するには セッションに。機能も使われています 参照 — コード内で他のモデルを参照する機能:

• コンパイル中 参照 ストレージ内のテーブルまたはビューへのターゲット ポインターに変換されます
• 参照 モデルの依存関係グラフを構築できます

すなわち ジンジャ DBT にほぼ無限の可能性を追加します。最も一般的に使用されるものは次のとおりです。

• If / else ステートメント - 分岐ステートメント
• for ループ
• 変数
• マクロ - マクロの作成

マテリアライゼーション: テーブル、ビュー、インクリメンタル

マテリアライゼーション戦略は、結果として得られるモデル データのセットをストレージに保存する方法です。

基本的には次のとおりです。

• テーブル - ストレージ内の物理テーブル
• ビュー - ストレージ内のビュー、仮想テーブル

さらに複雑な具体化戦略もあります。

• 増分 - (大きなファクト テーブルの) 増分読み込み。新しい行が追加され、変更された行が更新され、削除された行はクリアされます 
• 一時的 - モデルは直接具体化されませんが、他のモデルに CTE として参加します。
• 自分で追加できるその他の戦略

マテリアライゼーション戦略に加えて、次のような特定のストレージを最適化する機会もあります。

• スノーフレーク: 一時テーブル、マージ動作、テーブル クラスタリング、コピー許可、安全なビュー
• レッドシフト: Distkey、Sortkey (インターリーブ、複合)、Late Binding Views
• ビッグクエリー: テーブルのパーティショニングとクラスタリング、マージ動作、KMS 暗号化、ラベルとタグ
• スパーク: ファイル形式 (parquet、csv、json、orc、delta)、partition_by、clustered_by、バケット、incremental_strategy

現在、次のストレージがサポートされています。

• Postgres
• レッドシフト
• ビッグクエリー
• スノーフレーク
• プレスト（一部）
• スパーク（一部）
• Microsoft SQL Server (コミュニティアダプター)

モデルを改良してみましょう。

• 充填をインクリメンタルにしましょう (Incremental)
• Redshift にセグメンテーションとソートキーを追加しましょう

-- Конфигурация модели: 
-- Инкрементальное наполнение, уникальный ключ для обновления записей (unique_key)
-- Ключ сегментации (dist), ключ сортировки (sort)
{{
  config(
       materialized='incremental',
       unique_key='order_id',
       dist="customer_id",
       sort="order_date"
   )
}}
 
{% set payment_methods = ['credit_card', 'coupon', 'bank_transfer', 'gift_card'] %}
 
with orders as (
 
   select * from {{ ref('stg_orders') }}
   where 1=1
   {% if is_incremental() -%}
       -- Этот фильтр будет применен только для инкрементального запуска
       and order_date >= (select max(order_date) from {{ this }})
   {%- endif %} 
 
),
 
order_payments as (
 
   select * from {{ ref('order_payments') }}
 
),
 
final as (
 
   select
       orders.order_id,
       orders.customer_id,
       orders.order_date,
       orders.status,
       {% for payment_method in payment_methods -%}
       order_payments.{{payment_method}}_amount,
       {% endfor -%}
       order_payments.total_amount as amount
   from orders
       left join order_payments using (order_id)
 
)
 
select * from final

モデル依存関係グラフ

これも依存関係ツリーです。 DAG (有向非巡回グラフ) としても知られています。

DBT は、すべてのプロジェクト モデルの構成に基づいてグラフを構築します。つまり、モデル内の他のモデルへの ref() リンクです。グラフを使用すると、次のことが可能になります。

• モデルを正しい順序で実行する
• 店頭形成の並列化
• 任意のサブグラフの実行 

グラフの視覚化の例:

グラフの各ノードはモデルであり、グラフのエッジは式 ref によって指定されます。

データ品質とドキュメント

DBT では、モデル自体を生成するだけでなく、結果のデータセットに関する次のようなさまざまな仮定をテストできます。

• Null ではありません
• ユニークな
• 参照整合性 - 参照整合性 (たとえば、orders テーブルの customer_id は、customers テーブルの id に対応します)
• 許容可能な値のリストとの一致

たとえば、1 日、1 週間、1 か月前のインジケーターによる収益の % 偏差など、独自のテスト (カスタム データ テスト) を追加することができます。 SQL クエリとして定式化された仮定はすべてテストになる可能性があります。

このようにして、ウェアハウス ウィンドウ内のデータの望ましくない逸脱やエラーを検出できます。

ドキュメントの観点から見ると、DBT は、モデルレベル、さらには属性レベルでメタデータとコメントを追加、バージョン管理、配布するためのメカニズムを提供します。 

構成ファイル レベルでのテストとドキュメントの追加は次のようになります。

 - name: fct_orders
   description: This table has basic information about orders, as well as some derived facts based on payments
   columns:
     - name: order_id
       tests:
         - unique # проверка на уникальность значений
         - not_null # проверка на наличие null
       description: This is a unique identifier for an order
     - name: customer_id
       description: Foreign key to the customers table
       tests:
         - not_null
         - relationships: # проверка ссылочной целостности
             to: ref('dim_customers')
             field: customer_id
     - name: order_date
       description: Date (UTC) that the order was placed
     - name: status
       description: '{{ doc("orders_status") }}'
       tests:
         - accepted_values: # проверка на допустимые значения
             values: ['placed', 'shipped', 'completed', 'return_pending', 'returned']

生成された Web サイト上でこのドキュメントがどのように表示されるかは次のとおりです。

マクロとモジュール

DBT の目的は、SQL スクリプトのセットになることではなく、独自の変換を構築し、これらのモジュールを配布するための強力で機能豊富な手段をユーザーに提供することです。

マクロは、モデル内で関数として呼び出すことができる構成要素と式のセットです。マクロを使用すると、DRY (Don't Reply Yourself) エンジニアリング原則に従って、モデルとプロジェクト間で SQL を再利用できます。

マクロの例:

{% macro rename_category(column_name) %}
case
 when {{ column_name }} ilike  '%osx%' then 'osx'
 when {{ column_name }} ilike  '%android%' then 'android'
 when {{ column_name }} ilike  '%ios%' then 'ios'
 else 'other'
end as renamed_product
{% endmacro %}

そしてその使い方:

{% set column_name = 'product' %}
select
 product,
 {{ rename_category(column_name) }} -- вызов макроса
from my_table

DBT には、ユーザーが個々のモジュールやマクロを公開および再利用できるようにするパッケージ マネージャーが付属しています。

これは、次のようなライブラリをロードして使用できることを意味します。

• dbt_utils: 日付/時刻、サロゲート キー、スキーマ テスト、ピボット/アンピボットなどの操作
• 次のようなサービス用の既製のショーケース テンプレート 除雪機 и Stripe 
• 特定のデータ ストア用のライブラリ。 レッドシフト 
• ロギング — DBT 操作をログ記録するモジュール

パッケージの完全なリストは次の場所にあります。 DBTハブ.

さらに多くの機能

ここでは、チームと私がデータ ウェアハウスを構築するために使用する他のいくつかの興味深い機能と実装について説明します。 ウィリー.

実行環境の分離 DEV - TEST - PROD

同じ DWH クラスター内 (異なるスキーム内) であっても。たとえば、次の式を使用します。

with source as (
 
   select * from {{ source('salesforce', 'users') }}
   where 1=1
   {%- if target.name in ['dev', 'test', 'ci'] -%}           
       where timestamp >= dateadd(day, -3, current_date)   
   {%- endif -%}
 
)

このコードは文字通り次のように述べています: 環境向け 開発、テスト、CI 過去 3 日間のみデータを取得し、それ以降はデータを取得しません。つまり、これらの環境での実行ははるかに高速になり、必要なリソースも少なくなります。環境上で実行する場合 突く フィルタ条件は無視されます。

代替列エンコーディングを使用した実体化

Redshift は、個々の列ごとにデータ圧縮アルゴリズムを設定できる列指向 DBMS です。最適なアルゴリズムを選択すると、ディスク容量を 20 ～ 50% 削減できます。

大きい redshift.compress_table ANALYZE COMPRESSION コマンドを実行し、推奨される列エンコーディング アルゴリズム、指定されたセグメンテーション キー (dist_key) および並べ替えキー (sort_key) を使用して新しいテーブルを作成し、そこにデータを転送し、必要に応じて古いコピーを削除します。

マクロ署名:

{{ compress_table(schema, table,
                 drop_backup=False,
                 comprows=none|Integer,
                 sort_style=none|compound|interleaved,
                 sort_keys=none|List<String>,
                 dist_style=none|all|even,
                 dist_key=none|String) }}

モデルの実行のログを記録する

モデルの各実行にフックをアタッチできます。フックは、起動前またはモデルの作成完了直後に実行されます。

   pre-hook: "{{ logging.log_model_start_event() }}"
   post-hook: "{{ logging.log_model_end_event() }}"

ロギング モジュールを使用すると、必要なすべてのメタデータを別のテーブルに記録でき、後でボトルネックの監査と分析に使用できます。

Looker のログ データに基づくダッシュボードは次のようになります。

ストレージメンテナンスの自動化

UDF (ユーザー定義関数) など、使用するリポジトリの機能の拡張機能を使用する場合、これらの関数のバージョン管理、アクセス制御、新しいリリースの自動ロールアウトを DBT で行うと非常に便利です。

Python の UDF を使用して、ハッシュ、電子メール ドメイン、およびビットマスク デコードを計算します。

任意の実行環境 (dev、test、prod) で UDF を作成するマクロの例:

{% macro create_udf() -%}
 
 {% set sql %}
       CREATE OR REPLACE FUNCTION {{ target.schema }}.f_sha256(mes "varchar")
           RETURNS varchar
           LANGUAGE plpythonu
           STABLE
       AS $$  
           import hashlib
           return hashlib.sha256(mes).hexdigest()
       $$
       ;
 {% endset %}
  
 {% set table = run_query(sql) %}
 
{%- endmacro %}

Wheely では、PostgreSQL をベースにした Amazon Redshift を使用しています。 Redshift の場合、テーブルの統計を定期的に収集し、ディスク領域を解放することが重要です (それぞれ ANALYZE および VACUUM コマンド)。

これを行うために、redshift_maintenance マクロのコマンドが毎晩実行されます。

{% macro redshift_maintenance() %}
 
   {% set vacuumable_tables=run_query(vacuumable_tables_sql) %}
 
   {% for row in vacuumable_tables %}
       {% set message_prefix=loop.index ~ " of " ~ loop.length %}
 
       {%- set relation_to_vacuum = adapter.get_relation(
                                               database=row['table_database'],
                                               schema=row['table_schema'],
                                               identifier=row['table_name']
                                   ) -%}
       {% do run_query("commit") %}
 
       {% if relation_to_vacuum %}
           {% set start=modules.datetime.datetime.now() %}
           {{ dbt_utils.log_info(message_prefix ~ " Vacuuming " ~ relation_to_vacuum) }}
           {% do run_query("VACUUM " ~ relation_to_vacuum ~ " BOOST") %}
           {{ dbt_utils.log_info(message_prefix ~ " Analyzing " ~ relation_to_vacuum) }}
           {% do run_query("ANALYZE " ~ relation_to_vacuum) %}
           {% set end=modules.datetime.datetime.now() %}
           {% set total_seconds = (end - start).total_seconds() | round(2)  %}
           {{ dbt_utils.log_info(message_prefix ~ " Finished " ~ relation_to_vacuum ~ " in " ~ total_seconds ~ "s") }}
       {% else %}
           {{ dbt_utils.log_info(message_prefix ~ ' Skipping relation "' ~ row.values() | join ('"."') ~ '" as it does not exist') }}
       {% endif %}
 
   {% endfor %}
 
{% endmacro %}

DBTクラウド

DBTをサービス（マネージドサービス）として利用することが可能です。含まれるもの:

• プロジェクトとモデルを開発するための Web IDE
• ジョブの構成とスケジュール設定
• ログへの簡単かつ便利なアクセス
• プロジェクトのドキュメントが掲載されている Web サイト
• CI（継続的インテグレーション）を接続する

まとめ

DWH を準備して摂取することは、スムージーを飲むのと同じくらい楽しくて有益になります。 DBT は、Jinja、ユーザー拡張機能 (モジュール)、コンパイラー、エグゼキューター、およびパッケージ マネージャーで構成されます。これらの要素を組み合わせることで、データ ウェアハウスの完全な作業環境が得られます。現在、DWH 内の変革を管理するこれより良い方法はほとんどありません。

DBT の開発者が従う信念は次のように定式化されます。

• 複雑な分析ロジックを表現するには、GUI ではなくコードが最適な抽象化です
• データの操作には、ソフトウェア エンジニアリングのベスト プラクティスを適用する必要があります (ソフトウェア エンジニアリング)

• 重要なデータインフラストラクチャはオープンソースソフトウェアとしてユーザーコミュニティによって管理されるべきです
• 分析ツールだけでなく、コードもますますオープンソース コミュニティの所有物になるでしょう

これらの核となる信念は、現在 850 社を超える企業で使用されている製品を生み出し、将来作成される多くのエキサイティングな拡張機能の基礎を形成しています。

興味のある方のために、OTUS での公開レッスンの一環として数か月前に私が行った公開レッスンのビデオがあります - Amazon Redshift ストレージ用データ構築ツール.

DBT とデータ ウェアハウジングに加えて、OTUS プラットフォームのデータ エンジニア コースの一環として、同僚と私は他の多くの関連する現代的なトピックに関するクラスを教えています。

• ビッグデータ アプリケーションのアーキテクチャ概念
• Spark と Spark Streaming を使って練習する
• データソースをロードするための方法とツールを調べる
• DWH での分析ショーケースの構築
• NoSQL の概念: HBase、Cassandra、ElasticSearch
• モニタリングとオーケストレーションの原則 
• 最終プロジェクト: メンタリングサポートの下ですべてのスキルをまとめる

リンク：

1. DBT ドキュメント - はじめに — 公式ドキュメント
2. dbt とは正確には何ですか? — DBT の著者の一人によるレビュー記事 
3. Amazon Redshift ストレージ用データ構築ツール — YouTube、OTUS 公開レッスンの録画
4. グリーンプラムを知る — 次回の公開レッスンは15年2020月XNUMX日です
5. データエンジニアリングコース —オータス
6. 成熟した分析ワークフローの構築 — データと分析の未来を展望する
7. オープンソース分析の時代が来た — アナリティクスの進化とオープンソースの影響
8. dbtCloud を使用した継続的インテグレーションと自動ビルド テスト — DBT を使用した CI 構築の原則
9. DBT 入門チュートリアル — 独立した作業のための実践、段階的な指示
10. ジャッフルショップ — Github DBT チュートリアル — Github、教育プロジェクト コード

コースについて詳しくはこちらをご覧ください。

出所： habr.com