Spark Meetup Tokyo #2 (Spark+AI Summit EU 2019) 聴講メモ

# 概要

[Spark Meetup Tokyo #2 (Spark+AI Summit EU 2019)](https://spark-meetup-tokyo.connpass.com/event/153549/)

> Spark+AI Summit (SAIS) Europe 2019で発表があったSparkの最新開発状況，ユーザからのユースケース報告，関連OSSであるKoalas/MLflow/Delta Lakeなどに関する最新情報をお伝えします．

[前回分(Spark Meetup Tokyo #1)の聴講メモ](https://qiita.com/calderarie/items/07d674c858cbeffbbb8f)

# 各発表とそのメモ

各発表について、気になったところのメモを掲載します。

資料は公開予定とのことなので、公開され次第、リンクと内容について順次反映します。

## SPARK+AI Summit Europe 2019 セッションハイライト

- Keynote: Simplifying model Management with MLflow
    - MLflowの最新動向について
    - 機能アップデート(抜粋)
        - Autologging in Tensorflow & keras
    - 新規コンポーネント
        - MLflow Model Registry
    - Autologging
        - TensorflowやKerasで必要なParametersやMetricsを自動的にログへ残す
    - MLflow Model Registry
        - モデル管理機能
        - モデルに対して名前やコメント、タグを透けてバージョン管理できる
        - モデルごとに、状態遷移や履歴を管理できる
        - モデルに関するレビューやテストのワークフローを定義することが出来る
- Keynote: New Developments in the Open Source Ecosystem
    - Sparkエコシステムに関する最近の取り組み
        - Spark
        - Delta Lake
        - Koalas
    - Spark 3.0の改善ポイント
        - Adaptive Query Execution
            - 実行中にデータを見ながら動的にクエリプランを変更
        - Dynamic Partition pruning
            - 結合テーブルに対するフィルタ条件を基に、巨大なテーブルをフィルタ

- Downscaling: The Achilles heel of Autoscaling Apache Spark Clusters
    - クラウド上でのオートスケーリングについて
        - Sparkクラスタにおけるノードは、コンピュートと一時データの補完の役割
    - Upscaling
        - 負荷が高まったらノードを追加すればよい
    - 一方、Downscaling
        - 実行中のコンテナが存在しておらず、一時データ(shuffle/cache)が存在しない必要がある
        - Executorのフラグメンテーションへの対処として、ジョブスケジューリングを工夫
        - External Shuffle Serviceを有効にし、ShuffleデータをNFSへ

## [Koalas: Unifying Spark and pandas APIs](https://www.slideshare.net/ueshin/20191119-spark-meetup-tokyo-2)


- Koalasの開発状況
    - 公開直後は毎週リリースだったが、最近は隔週でリリース
    - pandasのAPIを優先的に実装

- 直近の新機能
    - plot系のメソッド
    - Spark関連のIO関係メソッド
        - to_XXXやread_XXXの対応(parquet, csv, json, ...)
    - SQLやcasheの対応
    - multi-indexのサポート

- 今度の対応
    - pandas APIの対応
        - rollingやexpanding (Window)
    - カテゴリ変数の対応
    - 時系列関係機能
    - パフォーマンス向上

## [Quick Overview of Upcoming Spark 3.0 + α](https://www.slideshare.net/maropu0804/quick-overview-of-upcoming-spark-30)

- Spark 次期リリース(3.0)と主だった変更点
    - 現状(2.4)との互換性が保証されていない
    - Adaptive Execution in Spark SQL
        - これまで、複雑なクエリでは、中間データのサイズの予測誤差で、最適な物理プランが選択できなかった
        - 物理プランを複数のQueryStageに分割し、各Stageで出力された統計情報を用いて物理プランの最適化を行う
    - Implement Dynamic Partition Pruning
        - Join処理において、片方テーブルのFilter条件を用いて、もう片方の不要なパーティション読み取りを除去するDynamic Filter導入
    - New format of EXPLAIN command
        - EXPLAINの出力を構造化するFORMATTEDが追加
    - Support Dataframe Cogroup via Pandas UDFs
        - Pandas UDFでcogroupな処理に対応
    - Implement join strategy hints
        - broadcast joinだけでなく、他のJoin方式に関するヒント情報の追加
            - Shuffle Hash Joins，Sort Merge Joins，Cartesian Product Joins，Broadcast Nested Loop Joins
    - Port test cases from PostgreSQL to Spark SQL
        - PostgreSQLとの互換モードの追加
- Powering Custom Apps at Facebook using Spark Script Transformation
    - Script Transformation
        - 外部プロセスを呼び出した処理が実行できる
        - UDFよりも柔軟で、(C++などにより)高速な処理が出来る
    - PyTorchを用いたモデル推論
        - 各クラスタにモデルファイルを置いておき、Script Transformationで推論処理

## [Delta アーキテクチャ](https://www.slideshare.net/JuanPauloGutierrez/delta-architecture)

- lambda architecture
    - ストリームとバッチに分割し、2つの処理方式管理
    - 2処理のバリデーションや各ロジックのメンテが課題
- Delta architecture
    - Structured streaming + Delta Lake
    - Delta Lakeを基に各処理

## [Project Hydrogenの最新情報](https://www.slideshare.net/ishizaki/sparktokyo2019novishizaki)
- Project Hydrogen
    - Spark上で分散AIフレームワークを効率よく学習推論を可能にする

- 分散学習の対応
    - これまでのSparkタスク(MapReduce)
        - Task独立に並列実行
    - 一方、分散学習
        - 並列実行する部分だけでなく、全体で実行する部分も
    - Barrier Execution Mode
        - 全てのタスクを同時に実行する
        - 故障が起きたら、
- アクセラレータ管理
    - GPUを各ExecutorやTaskに割当設定
    - プログラム上からアクセラレータの取得が可能に
- ストリームデータの推論
    - 推論時は、AIフレームワークとのデータ交換がボトルネックに
    - pandasUDF(内部でArrow)によって高速推論

## Our Story of tuning Spark Applications

- Spark Streaming Optimization
    - セール時にトラフィック増大によりストリーム処理遅延 → 後続処理も停滞
    - Executorを増やし、maxRatePerPartitionを調整

- Spark Report Engine Evolution
    - OLAPで処理できないレポートクエリ
    - SparkSQLへ変更したがCatalystバグによる実行できないケースがあったため、DatasetAPIへ以降

## [LT1: Koalasのココが良いよね](https://docs.google.com/presentation/d/1Kf1lYn4BGgqoxod-tYEdQqVrl2QF-0XUr0qlWOTeuIA/edit#slide=id.p)

- Pandasにはあるが、PySparkにない便利な機能
    - DataFrameがNotebook上で綺麗に表示される
    - is_unique
    - aggの挙動
    - cumsum
    - etc.

## LT2: 高階関数(Higher-Order Functions)をちょっと深掘りしてみた

- 高階関数
    - 引数として関数を受け取る
    - 戻り値として関数を返す
    - Spark2.4で追加
        - transform
        - filter
        - aggregate
        - exists
        - zip_with

- 整数配列の集計で比較
    - explodeとgroupBy
        - groupByによってShuffleが発生する
    - map
        - Deserializeが発生する
    - udf
        - 事前に処理内容を宣言する必要がある
        - 実行計画もシンプル
    - 高階関数
        - シンプルに記述可能
        - 実行計画もシンプル