DBTを使ったRedshiftへのデータロード

• Yelpは最近、消費者のデータ需要の増加に伴い、Redshiftにデータをより効率的にロードする方法を見直した
• この記事では、DBTをRedshift Spectrumとシームレスに組み合わせることで、Data LakeからRedshiftへデータを取り込み、実行時間を大幅に短縮し、データ品質の問題を解決し、開発者の生産性を向上させる方法を紹介する

出発点

• バッチデータをRedshiftにロードする方法は長年有効だったが、継続的に改善点を模索していた
• 主にSparkジョブを使ってS3データを読み込み、社内のKafkaベースのデータパイプラインに公開して、Data LakeとRedshiftの両方にデータを取り込んでいた
• しかし、いくつかの問題に直面し始めた:
  • パフォーマンス: 大規模データセット（1日あたり100万行超）で遅延が発生し始めた。主な原因は、upsert時に主キーの重複を防ぐためのテーブルスキャンだった
  • スキーマ変更: 多くのテーブルはAvroスキーマで構成されていた。スキーマ変更は、新しいAvroスキーマの生成と登録という複数段階のプロセスが必要なため、ときに複雑だった
  • バックフィル: データ修正に対するバックフィルのサポートが不十分だった。行をin-placeで更新する簡単な方法がなく、パーティション全体に修正済みデータを書き込む前に手動でデータを削除することが多かった
  • データ品質: Data LakeとRedshiftへ並列に書き込む方式には、2つのデータストア間でデータ型の違いなどによるデータ不一致が起きるリスクがあった

DBTでRedshiftロードを改善する

• データをより効率的に移動する方法を検討する中で、RedshiftからData Lakeのデータをクエリできるよう特別に構築されたツールであるAWS Redshift Spectrumを活用することにした
• Data Lakeのテーブルは通常もっとも新しいスキーマを持っていたため、RedshiftバッチのデータソースとしてS3ではなくData Lakeを使うことにした。これはデータ不一致の削減に役立つだけでなく、Data Lakeを単一の信頼できる情報源として扱うという自社のベストプラクティスにも合致していた
• 実装にあたり、Spectrumは定義済みスキーマを必要とするが、これはData Lakeテーブル向けにすでにGlueに存在していた。追加で必要だった設定は、Data Lakeテーブルを外部テーブルとして追加し、シンプルなSQLクエリでRedshiftからアクセスできるようにすることだけだった
• すでに他のデータセットでDBTの採用を進めていたが、パイプライン内でRedshift Spectrumクエリを取り込む用途にも最適な候補だった。DBTはデータ変換に優れており、モジュール化されバージョン管理されたSQLを書く運用を支援する
• SparkジョブがS3からRedshiftへ読み込む代わりに、DBTを使ってData LakeからRedshiftへデータを直接コピーした。DBTは再現性、柔軟性、データリネージのような代表的な利点を提供するだけでなく、前述のいくつかの問題の解決にも役立った

単純化されたスキーマ変更

• スキーマ変更を簡素化するために、DBTのon_schema_change設定引数を活用した。append_new_columnsに設定することで、入力データに存在しない列が削除されないようにした
• また、DBT contractを第2の保護層として用い、書き込まれるデータがモデル設定と一致していることを確認した

手作業の少ないバックフィル

• DBTによってバックフィルも大幅に容易になった。pre_hook設定引数を使うことで、モデルの直前に実行するクエリを指定できる
• これにより、書き込まれるパーティションのデータをより自動的に削除できるようになった
• その結果、べき等性を保証できるようになり、古いデータが削除されないことを心配せずにバックフィルを実行できるようになった

データの重複排除

• 重複行に対処するため、SQLに重複排除レイヤーを追加し、DBTテストで検証した
• DBTには組み込みのunique列テストがあるが、テーブル全体のスキャンが必要なため、大きなテーブルでは現実的ではない
• 代わりに、dbt_expectationsパッケージのexpect_column_values_to_be_unique関数を使用した。これにより、直近で書き込まれた行だけをスキャンする行条件を指定できる

パフォーマンス向上

• もっとも目立った成果は、特に最大の問題を抱えていたRedshiftデータセットにおけるパフォーマンス改善だった:
  • 書き込みには約2時間かかっていたが、現在は通常わずか10分で実行できる
  • 以前は月あたり最大6時間の遅延があったが、現在は遅延がなくなった。これはオンコールでのインシデント対応の負担を大きく軽減した
  • スキーマアップグレードは以前はより長い多段階プロセスだったが、現在は数時間で完了する3段階プロセスに改善された

より良いデータ整合性

• データフローの分岐をなくすことで、異なるデータストア間でデータが食い違わないという確信が高まった
• Redshiftに入るすべてのデータはまずData Lakeを通過する必要があるため、Data Lakeを単一の真実の源として維持できることをより確実にした

結論

• この移行の成功を受けて、約12の他のデータセットにもこれらの変更を適用し、全体として同様の利点を確認した
• AWS Redshift SpectrumやDBTのようなツールを活用し、進化するデータ要件に合わせてインフラを適応させることで、ユーザーとステークホルダーにより大きな価値を提供している