Google Cloud Platform Japan 公式ブログ: Cloud Dataflow パイプラインのモニタリングと高速化

Cloud Dataflow パイプラインのモニタリングと高速化

2016年6月20日月曜日

* この投稿は米国時間 6 月 8 日、Software Engineer である Robert Burke によって投稿されたもの（投稿はこちら）の抄訳です。注記 : この投稿で紹介する機能は現在リリースが進められているところです。近いうちに Google Developers Console に追加されるでしょう。Google Cloud DataflowCloud Dataflow

Dataflow Monitoring Interface のジョブサマリーには、ジョブの Elapsed Time（経過時間）が秒単位で表示されます。上記の場合、予想とは違って約 20 分となっています。

このサマリーは、Total Worker Time（総ワーカー時間 : ジョブが消費した計算時間をワーカーの稼働秒数で表したもの）も表示します。Total Worker Time は Elapsed Time よりはるかに長くなることがあります。複数のワーカーが並列に稼働するケースがあるからです。

上記のジョブの場合、Total Worker Time が 1 時間近くに達していることがわかります。しかし、どうしてそこまで時間がかかっているのかはわかりません。

一見すべてがうまくいっているようですが、実はそうではありません。ExtractUserScore に異常に長い時間がかかっていたのです。

複合ステップである ExtractUserScore の実行時間は、ネストされたステップすべての実行時間の合計です。どこに問題があるのか、掘り下げてみます。

上のグラフを見ると、GroupByKey に多くの時間が費やされていることがわかります。このグラフは、パイプラインが値をグループ化し、その合計を計算することを示しています。どこを見ればよいかわかったので、コードに切り替えます。

...
.apply(GroupByKey.<String, Integer>create())
.apply( 
       ParDo.named("MySum")      
               .of(          
                     new DoFn<KV<String, Iterable<Integer>>, KV<String, Integer>>() {              
                            @Override              
                            public void processElement(ProcessContext c) {                
                                String key = c.element().getKey();                
                                int summedValue = 0;                
                                for (int i : c.element().getValue()) {                  
                                  summedValue += i;                
                                }                
                                c.output(KV.of(key, summedValue));             
                            }         
                      }))
...

Beam プログラミングモデルの結合プリミティブではなく、カスタムロジックを使用しています。ひょっとすると、すべてのスコアが合計される前にランダムに並べ替えられているのかもしれません。グラフに戻って調べてみましょう。

案の定、そうでした。わずか 8,000 の要素に対して 10 GB 近くのデータがランダムに並べ替えられています。しかし幸いなことに、これなら問題解決は簡単です。合計計算は結合的かつ可換なので、Combine を使ってパイプラインをより的確に表現できます。

パイプラインのセマンティクスをより厳密に記述することで、Cloud Dataflow のパフォーマンスが向上します。この場合、パイプラインは Combine Optimization（結合最適化）を実行します。これは、各ワーカーが並べ替えの前に、値の一部を合計することを意味します。これによってワーカー間で受け渡されるデータ量が大幅に少なくなり、並べ替えが著しく高速化します。

...
.apply(GroupByKey.<String, Integer>create())
.apply("MySum",       
           Combine.<String,Integer>groupedValues(            
                     new SerializableFunction<Iterable<Integer>, Integer>() {                
                           @Override                
                           public Integer apply(Iterable<Integer> values) {                  
                               int summedValue = 0;                  
                               for (int i : values) {                    
                                  summedValue += i;                  
                               }                  
                               return summedValue;                
                           }            
                    }))
...

上のコードでは、Combine.groupedValues を使用するように変更しています。パイプラインを再び実行してみましょう。

ご覧のとおり、はるかに改善されました。パイプラインの実行にかかった時間（Elapsed Time）は 7 分を切っています。データの読み取り時間と解析時間も短縮されました。パイプラインがすべてのスコアのランダムな並べ替えにあまり時間を費やさなくなったからです。

この並べ替えは、前のステップであるデータの読み取りや解析のパフォーマンスにも悪影響を与えていました。1 時間近くを示していた Total Worker Time にはこうした改善が反映され、17 分弱まで減少しています。

ランダムに並べ替えられたデータの量は、今回は前回の 1 万分の 1 となり、1 MB に満たなくなっています。Combine を使用することでパフォーマンスがどれだけ向上するかはデータセットによって異なりますが、データを適切に結合できれば、その効果は絶大です。

結合的かつ可換な複数の部分に分割できる演算はすべて、Combine で表現可能です。たとえば平均は、総数と総量を別々にカウントし、出力が必要なときに、総量を総数で割ることで計算できます。Cloud Dataflow SDK は一般的な統計的結合の関数を提供しており、この例のカスタムコードは Sum.integersPerKey() を使って記述できます。