Data Engineering Meetup #1 勉強会メモ

# 概要

[Data Engineering Meetup #1](https://dem.connpass.com/event/120994/)

>Data Engineering Meetup は、データエンジニアリング（データの収集・管理・処理・可視化など）に関する技術の情報交換・共有を行うためのミートアップイベントです。


# 発表

ところどころ抜けや間違いがあると思うので参考程度に

## Spark 2.4 & 3.0 - What's next -

[発表資料](https://www.slideshare.net/nttdata-tech/apache-spark-24-and-30-whats-next)

### 演者
株式会社エヌ・ティ・ティ・データ 猿田 浩輔 様

### 内容

- Sparkの変遷
    - 最新ver 2.4
    - 今年リリース予定ver 3.0
    - 直近では、RDDからSparkSQLへの処理系へ移行し、パフォーマンスと利便性が向上
        - 宣言的なAPIやSQLで処理が書ける
        - オプティマイザによる最適化の恩恵を受けられる
        - 開発言語による性能差が出にくい
        - MLLibなどのライブラリもDataFrameを利用したものが整備されてきた
- kubernetesサポート
    - 2.3からサポートされた
    - 2.4対応
        - PySparkとSparkR対応
        - クライアントモードによるインタラクティブな利用
    - 3.0対応
        - Kerberos認証サポート
        - ダイナミックリソースアロケーション
- Project Hydorogen
    - Sparkにおけるディープラーニングを活用したユースケースをカバーするサブプロジェクト
        - Barrier Execution Mode
        - Accelerator Aware Scheduling
        - Optimized Data Exchange

    - BarrierExecution Mode (2.4で基本対応)
        - ディープラーニングは、複数タスクの協調動作が必要とされる
        - 従来のSparkのタスクスケジューリングでは難しいため、新しいスケジューリング機能

    - Accelerator Aware Scheduling (3.0対応)
        - GPUのサポート
 
- Spark Graph(3.0対応)
    - 既存のグラフ処理ライブラリであるGraphXやGraphFrameの課題を解決するライブラリ
        - GraphX
            - あまりメンテナンスされていない
            - RDDベースでScalaAPIしか提供されていない
        - GraphFrames
            - SparkPackagesで提供されるサードパーティパッケージ
            - DataFrameベース
            - 単純なグラフマッチングしか行えない
            - エッジやノードのセマンティックが弱く、単純なグラフマッチングしか行えない
    - SparkGraph
        - GraphFramesでは、エッジやノードの型を指定できず、グラフ形状によるマッチングのみ 
        - 型が指定できるProperty Graphというデータモデルを採用
        - Cypherによるクエリで、型や属性を用いたグラフマッチングが可能

## 持続可能なデータ基盤のためのデータの多様性に対する取り組み

[発表資料](https://www.slideshare.net/cyberagent/dem-20190319zenmyo)

### 演者
株式会社サイバーエージェント 善明 晃由 様

### 内容

- データエンジニアリングは技術的負債が生まれやすい
    - 専門性が高く、採用が難しい（人がいない） 
    - 規模が大きく、アップデートにかかるコストが大きい
    - 技術の進歩が速い

- 問題意識
    - 各サービスが独立して案件を進めるため、サイロ化
    - 各案件に十分なリソースを避けず、担当者にフルスタックなスキルが求められる
    - 開発から運用まで一連をやる必要があり、専門性が深められない
    - 変化が速いために、技術的負債につながりやすい
 
- 方針
    - 各システムの共通部分をプロダクト化を促進
    - 役割を分離して、専門性を高める

- HBaseのスキーマ設計
    - Hiveの集計結果をHBaseに書き込み、結果を可視化
    - HBaseは、キーによってパフォーマンスが大きく変わるので、スキーマ設計が重要
    - 用途によって有効なスキーマが異なる
        - PVやUUの集計
        - ページごとのPVやレスポンスタイム
        - 広告の効果測定 
    - HiveからHBaseのKeyValueに変換する汎用的な設計モデルツールを作成

    - やることが、対応スキーマの拡張と問い合わせ処理の最適化のみに

- WebUIでのレポート
    - 元となるデータは表形式
    - 列は2種類
        - ディメンジョンとメトリック
    - Grafanaで出力できる統一的なフォーマット化

    - やることが、フロントエンドでの可視化やUI改善のみに

## Deep Dive into Spark SQL with Advanced Performance Tuning

[発表資料](https://www.slideshare.net/ueshin/20190319-deep-dive-into-spark-sql-with-advanced-performance-tuning)

### 演者
Databricks Inc. 上新 卓也 様

### 内容

- SparkSQLの内部詳細
    - クエリからRDDを生成する
    - df.explainでプランを生成、解釈し、パフォーマンスチューニングを実施する

- Metadata Catalog
    - パーティションのメタデータ取得のコストを下げる 
        - Hive metastoreのアップグレード
        - cardinalityの高いパーティションカラムを避ける
        - 取得時にパーティションの絞り込みを入れる

- Cache Manager
    - 始めて利用するときにキャッシュデータ作成
    - 以降、プランが一致した場合に、キャッシュデータと置き換える

    - ディスクに書き出される場合があり、必要以上にキャッシュしないようにする
   
- Optimizer
    - 生成されたロジカルプランを最適化 
    - ヒューリスティックとコストベースでプランの書き換え

    - 独自にOptimizerやPlannnerRuleを組み込むことができる
  
- Plannner
    - ロジカルプランをフィジカルプランへ
    - 各プランのコストに基づいて最適なフィジカルプラン選択

    - 結合時のbroadcast joinとshuffle joinの選択(前者が早い)
        - spark.sql.autoBroadCastJoinThresholdでbroadcast joinをする閾値を設定
        - 統計情報を細心に保つ
        - クエリにbroadcastJoinヒントを組み込み、明示的に選択
    - Join条件に少なくとも=条件を１つ入れることで、計算量削減

- Query Excecution
    - Memory Managerのメモリ管理
        - spark.executor.memoryとspark.memory.frationに多少余裕を持たせて設定する
        - Spkarの監視外のメモリのため

        - spark.memory.offHeap.enabledとspark.memory.offHeap.sizeを設定して、オフヒープを有効化
        - そのぶんspark.executor.memoryを減らしてよい

    - Code GeneratorのフィジカルプランからJavaコードの生成
        - spark.sql.codegen.hugeMethodLimitを設定
        - バイトコードのサイズが8kバイトを超える大きなメソッドはJITコンパイルできず、インタプリタで実行するため遅くなる
        - 設定を8kにするとよい？

## Improving Spark SQL Performance

[発表資料](https://speakerdeck.com/line_developers/improving-spark-sql-performance)

(Hadoop / Spark Conferenceから参照したので違う資料かも)

### 演者
LINE株式会社 吉田 啓二 様

### 内容

- OASIS
    - 内製のBIツール
    -　1日200人以上、700以上のSparkジョブ

    - 課題：遅いクエリが存在する
       - エンドユーザのクエリの改善によって、パフォーマンスを解決することが難しい
       - プラットフォーム側で性能改善を頑張る

- 結合方式の最適化
    - sort merge join
        - 結合するテーブルそれぞれでシャッフルされる
    - broadcast hash join
        - １つのテーブルのみをシャッフルするため、早い
        - シャッフルするテーブルがメモリ上に乗せる
        - spark.sql.broadcashhashjoinThresholdで指定される(デフォルト10MB)
    - できる限りbroadcast hash joinが選択されるようにする

- ケース1
    - コスト見積もりが正しく行えていない
        - spark.sql(query).queryExecution.otimizedPlan.stats.sizeInBytes
    - データ追加時に、統計情報を取得して、正しく見積もりできるようにする
        - sql('ANALYZE TABLE [db_name.]table_name COMPUTE STATISTICS NOSCAN')

- ケース2
    - ファイルをLOAD DATAで参照しているhiveテーブルの統計情報がとれない
        - hive.stats.autogetherの仕様 
    - broadcastヒント
        - 結合時にbroadcastを選択することができる
        - クエリに明示的に入れる必要があるため、ここでは却下
  
    - spark.experimental.extraOptimizationsで独自の最適化ルールを追加

- cost-based optimizationについて
    - Spark 2.2.0~
    - デフォルトはルールベース
    - 設定値でコストベースに変えることが出来る
        - spark.sql.cbo.enabled = true
        - spark.sql.cbo.joinReorder.enabled = true 
    - broadcast hash joinをより精緻に選択される
        - ルールベースでは、テーブル単位の統計情報を参照している
        - where句などの絞り込み後のカラム単位の統計情報を参照する
    - 結合順を最適化される
        - 小さいテーブルから結合して、シャッフルのコストが少なくなる