AIチームのためのデータエンジニアリング

# はじめに
* データエンジニアリングの基本知識の内、AIチームでも特に使いそうな要素をかいつまんで簡単にまとめました。
* （追記）その後、Google Cloud Professional Data Engineer取得に向けて勉強し直したときのメモ: [GCPでデータエンジニアリング入門](https://qiita.com/nokoxxx1212/items/21706a7052714ff109e4)

# 参考書籍
* [Google Cloudではじめる実践データエンジニアリング入門](https://www.amazon.co.jp/Google-Cloud%E3%81%A7%E3%81%AF%E3%81%98%E3%82%81%E3%82%8B%E5%AE%9F%E8%B7%B5%E3%83%87%E3%83%BC%E3%82%BF%E3%82%A8%E3%83%B3%E3%82%B8%E3%83%8B%E3%82%A2%E3%83%AA%E3%83%B3%E3%82%B0%E5%85%A5%E9%96%80-%E6%A5%AD%E5%8B%99%E3%81%A7%E4%BD%BF%E3%81%88%E3%82%8B%E3%83%87%E3%83%BC%E3%82%BF%E5%9F%BA%E7%9B%A4%E6%A7%8B%E7%AF%89-%E4%B8%8B%E7%94%B0-%E5%80%AB%E5%A4%A7/dp/429711948X)
* [ビッグデータを支える技術](https://www.amazon.co.jp/%E5%A2%97%E8%A3%9C%E6%94%B9%E8%A8%82-%E3%83%93%E3%83%83%E3%82%B0%E3%83%87%E3%83%BC%E3%82%BF%E3%82%92%E6%94%AF%E3%81%88%E3%82%8B%E6%8A%80%E8%A1%93-%E2%80%94%E2%80%94%E3%83%A9%E3%83%83%E3%83%97%E3%83%88%E3%83%83%E3%83%971%E5%8F%B0%E3%81%A7%E5%AD%A6%E3%81%B6%E3%83%87%E3%83%BC%E3%82%BF%E5%9F%BA%E7%9B%A4%E3%81%AE%E3%81%97%E3%81%8F%E3%81%BF-PRESS-plus/dp/4297119528/ref=pd_lpo_1?pd_rd_i=4297119528&psc=1)

# データエンジニアリングとは
* データ活用のためのインフラストラクチャを設計・構築・運用するための技術
* （上記の参考書籍に書かれているような内容）

# データエンジニアリングの必要性について
* MLシステムを構築する際に、データエンジニアリングを全く考慮しないと、下記のような課題が発生し得ります。
![スクリーンショット 2021-06-20 23.22.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/eae8a3c5-8b42-52fa-58ff-b892f0efd100.png)

# ①DWH
## DWHとは
* 分析のためにデータを格納しておくためのシステム
* 使いどころ
  * 集約することで、データのサイロ化を防ぐ（Dx条件の一つ）
  * データベースとの違いは、「データ分析が目的であること」「膨大なデータを高速に処理する必要があること」
* （補足）データレイク・データマートとは
    * データレイク: 未加工のデータを収集して蓄積するための場所。非構造化データ含めて。
    * [データ分析基盤の基本と構築のポイント](https://data.nifcloud.com/blog/data-analysis_basic_point/)から引用

![スクリーンショット 2021-06-19 22.28.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/1618640e-711c-5293-002d-17ad2f22e451.png)

* （参考）列指向データベース
    * データ分析では、しばしば一部のカラムだけが集計の対象になる
    * -> データをあらかじめカラム単位にまとめておくことで、必要なカラムだけを読み込むことでディスクI/Oを減らす。
    * -> 同じカラムには、しばしば同じようなデータが並ぶので、圧縮できる
* 設計のポイント
  * データ保存量の拡張手段（制約がないか）
  * データの取り出し手段

## 実用イメージ
* [Building an ETL data pipeline: GCS-Dataprep-BigQuery](https://rspraneethkumar.medium.com/building-an-etl-data-pipeline-gcs-bigquery-dataprep-5a5dbd3eba24)から引用

![スクリーンショット 2021-06-19 22.16.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/f3469331-5b63-0c85-ee0e-9863b83aa6b9.png)

## 【GCP】BigQuery
* BigQuery は、Google Cloud のペタバイト規模の費用対効果に優れたフルマネージド型の分析データウェアハウスであり、膨大な量のデータに対してほぼリアルタイムで分析を行うことができる
* BQの仕組み

![スクリーンショット 2021-06-21 14.51.30.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/27d7550b-c20b-73d4-0293-798748b69157.png)

* RDBとの使い分け
  * データマート用途であれば、1千万レコード/5GB（1レコード/500バイト仮定）くらいならRDBでも問題ないかも？
* GCSデータのクエリ
    * BigQuery では、次の形式の Cloud Storage データのクエリがサポートされています。
    * -> カンマ区切り値（CSV）,JSON（改行区切り）,Avro,ORC,Parquet,Datastoreエクスポート,Firestoreエクスポート
* データの取り込み方法
    * バルクロード: BigQuery Data Transfer Service, など
    * ストリーミング挿入: Fluentd, Dataflowのコネクタ, StriimなどのCDCソリューション（MySQLなどと同期）
    * フェデレーション: 外部テーブルを指定 
* コスト
  * マルチテナント方式であるため、事前のプロビジョニングのコストが発生しない
      * コンピュートまたはストレージのどちらか大きいほうのピークに合わせてサイジング調整する必要などがない
  * クエリでスキャンした容量に対して課金
* BigQueryML
    * SQLを用いて、機械学習のステップを完結できる
        * BQが予測リクエストをSQL形式で受付け、推論結果をサービングできる
    * サポートしているモデル
        * 線形回帰（予測）
        * 2項ロジスティック回帰（分類）
        * 多項ロジスティック回帰（分類）
        * K平均法クラスタリング（データセグメンテーション）
        * 行列分解（商品のレコメンデーションシステムの作成）
        * 時系列（時系列予測）
        * ブーストツリー（XGBoostベースの分類モデルと回帰モデルの作成）
        * ディープ ニューラル ネットワーク（DNN）（分類、回帰）
        * **AutoML Tables**: 特徴量エンジニアリングやモデル選択を行わずに、最適なモデルを作成する
        * TensorFlow モデルのインポート

# ②データ処理-Spark
## Sparkとは
* ビッグデータ分析に最適なオープンソースの分散処理システム。データ処理で最大のオープンソースプロジェクト。
* Netflix、Yahoo、eBay などのインターネット大手も、Spark を大規模にデプロイし、8000 を超えるノードのクラスターで、複数のペタバイトデータをまとめて処理しています
* [Cluster Mode Overview](https://spark.apache.org/docs/latest/cluster-overview.html)から引用

![スクリーンショット 2021-06-20 1.00.13.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/1af94011-063a-18a5-6f15-a96f3e0b3dda.png)

## 実用イメージ
* [Using Apache Spark with TensorFlow on Google Cloud Platform](https://cloud.google.com/blog/products/gcp/using-apache-spark-with-tensorflow-on-google-cloud-platform)から引用

![スクリーンショット 2021-06-20 0.54.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/085db450-2bb0-8bd4-3084-6c532df7a889.png)

## 【GCP】DataProc
* マネージドHadoop/Spark
* メモリを追加してPresto, GPUを追加してSparkで機械学習もできる
* データソース
  * ファイルパスの接頭辞をhdfs://からgs://へ変更するだけでGCSにアクセスすることができる
  * BQのストレージをDataProcから直接触ることもできる（BigQueryStorageAPIの利用）
  * DataProcでGCS上にHiveパーティションデータを作る→それをBQで直接触る、などもできる
* 90sで起動する
* コスト
    * プリエンプティブル VM で実行することによって費用をさらに節約できる
    * -> プリエンプティブルワーカーは、他のタスクで Google Cloud が必要とする場合に、再利用（クラスタから削除）されます
* サンプルコード
  * [チュートリアル](https://cloud.google.com/dataproc/docs/tutorials/gcs-connector-spark-tutorial#python_1)

# ③データ処理-Beam
## Beamとは
* Implement batch and streaming data processing jobs that run on any execution engine
![スクリーンショット 2021-06-20 8.21.33.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/2343b95e-c607-0ee7-ac5c-2c9a7fcdf487.png)
* データ読み込み用のコネクタは、BQ,GCS,PubSub,Kafka,Kinesis,SQS,Cassendra,JDBCなどに対応
* Apache Beam で記述したデータ処理のジョブはさまざまなビッグデータ基盤で動作させることができる。Spark,Hadoop MapReduce,Dataflow,,,
* SpotifyはリアルタイムレコメンドにApacheBeamを利用。ピークで800万イベント/sのストレーミングイベントを処理
* 複数の TFX ライブラリではタスクの実行に Beam が使用されているため、コンピュートクラスタでの高度なスケーラビリティが可能
* （参考）ラムダアーキテクチャとカッパアーキテクチャ
  * ストリーム処理で対応すべき課題として「間違った結果をどのように修正するか」「遅れてくるデータの扱い」
  * ラムダアーキテクチャ: バッチ処理とストリーム処理の両方を組み合わせた
      * 最新のデータは、ストリーム処理から生成されたリアルタイムビューを参照し、それより前はバッチ処理されたバッチビューを参照する
  * カッパアーキテクチャ: スピードレイヤとバッチレイヤの処理を同じツールで行う
      * やり直しに備えて、メッセージブローカのデータ保持期間を十分に長くする

## 実用イメージ
* [Machine Learning on Streaming Data using Kafka, Beam, and TensorFlow](https://www.slideshare.net/ConfluentInc/machine-learning-on-streaming-data-using-kafka-beam-and-tensorflow-mikhail-chrestkha-google-cloud-stephane-maarek-datacumulus-kafka-summit-nyc-2019)から引用
    * ストリーミング処理として、キューから受け取って処理をして、BQやMLサービング環境に投げる、など
![スクリーンショット 2021-06-20 1.24.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/272f672d-cbc3-977d-c990-4ac213c5b749.png)

## 【GCP】Dataflow
* [BigQueryユーザのためのCloud Dataflow入門](https://docs.google.com/presentation/d/1cHec9y46gegcbp94TrUMdF2OWOsRDphwJiYUqGsxTUc/edit?usp=sharing)が分かりやすい
* リアルタイムに整形・集計などのデータ処理をするパイプラインを構築する
* 他GCPサービスとの連携
  * Pub/Subで収集→Dataflowで処理→BQで蓄積、など
  * S3に書き出すこともできる
  * **リアルタイムなアーキテクチャは、GCPだと以下2パターンが多い**
      * DataflowでGCSに保管してフェデレーション
      * BQのストリーミング挿入機能
* サンプルコード
  * [チュートリアル](https://cloud.google.com/dataflow/docs/samples/molecules-walkthrough)

# ④ワークフロー管理
## ワークフロー管理とは
* 処理の依存関係を制御、処理のタイミングを適切にコントロールする
* データパイプライン（データがどこから来てどの処理を経てどこに行くのか）を管理するツールであり、多くはDAG（有向非循環グラフ）で一連の処理を記述する
* メリット
  * 依存関係が複雑になりがちな処理フローを管理できる
  * 実装を共通化できる
  * 実験環境から本番環境への移行がしやすい
  * エラー通知、リカバリーの支援などの運用管理面
* サンプルコード
    * [チュートリアル](https://airflow.apache.org/docs/apache-airflow/stable/tutorial.html)

## 実用イメージ
* [Machine Learning in Production using Apache Airflow](https://towardsdatascience.com/machine-learning-in-production-using-apache-airflow-91d25a4d8152)から引用
    * 図は前処理のみ分岐しているが、前処理は共通だがtrainを分岐させたいパターンなども多い
![スクリーンショット 2021-06-20 10.29.35.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/f33a0f8b-f5df-77d0-063a-82f6f5f5938d.png)

## 【GCP】CloudComposer
* Airflowに必要なコンポーネントが自動で起動してくれる
  * GKE
  * GCS
  * CloudLogging
  * CloudMonitoring
  * CloudSQL（バックエンドDB）

# ⑤特徴量ストア
## 特徴量ストアとは
* 特徴量の読み書きに特化したデータストア
* 「組織内の誰もが特徴量にアクセスできるように」「ビッグデータと機械学習の世界を隔てる境界線」
* 特徴
    * オンラインとオフラインという性質の異なる2つのシステムにデータを供給
        * オンラインはNoSQLで高頻度の読み書きに耐えられるように。オフラインは長期的な保存のためオブジェクトストレージ、など
    * バージョニング: 変更履歴を記録
    * タイムトラベル: 任意の時点にデータを巻き戻す
    * データリネージ: データの依存関係追跡
    * データの検証: スキーマを定義

## 実用イメージ
* [Feature Stores: Components of a Data Science Factory](https://neptune.ai/blog/feature-stores-components-of-a-data-science-factory-guide)から引用
![スクリーンショット 2021-06-20 22.22.02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/be3768c8-7529-0b69-5f21-689e92e838f5.png)
![スクリーンショット 2021-06-20 22.22.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/95a2cc96-7e00-eac2-3a9f-29ba18aad012.png)

## 【GCP】Vertex Feature Store
* ML特徴の整理、保存、サービングで使用される一元化されたリポジトリを提供
* 検索機能とフィルタ機能により、他のユーザーも既存の特徴を簡単に見つけて再利用できる
* オンライン予測をタイムリーに行うために低レイテンシのデータ サービングインフラストラクチャを構築
* 本番環境で使用する特徴データの分布が、モデルのトレーニングに使用された特徴データの分布と異なることを回避

# 参考
* [クラウドDWH（データウェアハウス）って何？](https://www.topgate.co.jp/aws-azure-gcp-dwh#dwh)
* [BigQuery ML とは](https://cloud.google.com/bigquery-ml/docs/introduction)
* [BigQueryユーザのためのCloud Dataflow入門](https://docs.google.com/presentation/d/1cHec9y46gegcbp94TrUMdF2OWOsRDphwJiYUqGsxTUc/edit#slide=id.p)
* [beam](https://beam.apache.org)
* [Vertex Feature Store とは](https://cloud.google.com/vertex-ai/docs/featurestore/overview)