Dataflowでプロファイリングとかダンプとか見てみる

Dataflowのパフォーマンスが気になった時のメモです。
Cloud monitoring、Cloud Profiler、メモリダンプがDataflowで使えます。


# Cloud monitoring
Cloud monitoringでJVMのメモリ使用量などをモニタリングする事が出来ます。


## わかること
* JVMのメモリ使用量
* スレッド数
* その他[agent metrics](https://cloud.google.com/monitoring/api/metrics_agent#agent)

## 試してみる
デフォルトでは無効なので、[--experiments=enable_stackdriver_agent_metrics](https://cloud.google.com/dataflow/docs/guides/using-stackdriver-monitoring#receive_worker_vm_metrics_from_the_agent)で、パイプラインの起動時（もしくはテンプレートのビルド時）に有効にします（※）。
※ experimentsとなっていますが何か懸念事項があるかは不明です‥


[Google提供のテンプレート](https://github.com/GoogleCloudPlatform/DataflowTemplates)の[PubSubToText](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/master/src/main/java/com/google/cloud/teleport/templates/PubsubToText.java)で試してみます。下から二行目が追加した部分です。

```
 mvn compile exec:java \
 -Dexec.mainClass=com.google.cloud.teleport.templates.PubSubToText \
 -Dexec.cleanupDaemonThreads=false \
 -Dexec.args=" \
 --project=${PROJECT_ID} \
 --stagingLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/staging \
 --tempLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/temp \
 --runner=DataflowRunner \
 --windowDuration=2m \
 --numShards=1 \
 --inputTopic=projects/${PROJECT_ID}/topics/windowed-files \
 --outputDirectory=gs://${PROJECT_ID}/temp/ \
 --outputFilenamePrefix=windowed-file \
 --experiments=enable_stackdriver_agent_metrics  \
 --outputFilenameSuffix=.txt"
```

ジョブ起動後、Cloud Monitoringで指標を確認することが出来ます。デフォルトではインスタンス毎ですが、ジョブ名でGroup Byすることも出来ます。

![Cloud Monitoring例](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/311203/417e5748-45d5-ba12-f271-64ad6ae4dd3b.png)



なお、[Word Count](https://beam.apache.org/get-started/wordcount-example/)でも試したのですが、こちらはMonitoringに表示されませんでした‥（原因は未調査。実行時間が短い？）


# Stackdriver Profiler
Googleの中の人が[Meidum](https://medium.com/google-cloud/profiling-dataflow-pipelines-ddbbef07761d0)で紹介している方法です。


## わかること
* CPU時間の[フレームグラフ](https://cloud.google.com/profiler/docs/concepts-flame?hl=ja)
* 実時間のフレームグラフ

Stackdriver ProfilerはJavaだと[ヒープも取れそう](https://cloud.google.com/profiler/docs/concepts-profiling?hl=ja#types_of_profiling_available)ですが、Dataflowで取れるかはわかりませんでした‥


## 試してみる
profilingAgentConfiguration='{ \"APICurated\": true }'オプションを有効にする必要があります。
なお、Cloud Monitoringと同時に使うことも出来ます。

```
 * mvn compile exec:java \
 -Dexec.mainClass=com.google.cloud.teleport.templates.PubSubToText \
 -Dexec.cleanupDaemonThreads=false \
 -Dexec.args=" \
 --project=${PROJECT_ID} \
 --stagingLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/staging \
 --tempLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/temp \
 --runner=DataflowRunner \
 --windowDuration=2m \
 --numShards=1 \
 --inputTopic=projects/${PROJECT_ID}/topics/windowed-files \
 --outputDirectory=gs://${PROJECT_ID}/temp/ \
 --outputFilenamePrefix=windowed-file \
 --experiments=enable_stackdriver_agent_metrics  \
 --outputFilenameSuffix=.txt"
```

![経過時間](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/311203/0bb96e28-a3fe-7c8e-b7f6-50fa7f473964.png)

![CPU時間](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/311203/4f2cc173-8979-9ba6-8b86-e362437c9c3e.png)


グラフの見方はStackdriver Profilerの[クイックスタート](https://cloud.google.com/profiler/docs/quickstart?hl=ja)や、[フレームグラフの説明](https://cloud.google.com/profiler/docs/concepts-flame?hl=ja)を参考にしてください。


# メモリダンプ
[Google Cloud Platform Community](https://cloud.google.com/community/)で[紹介](https://github.com/GoogleCloudPlatform/community/blob/master/tutorials/dataflow-debug-oom-conditions/index.md)されていますが、定番のメモリダンプも使えます。
上述の2つの方法と違い、ジョブ全体ではなく、インスタンス単位で見ることになります。

ダンプを取る方法は

* OOM時にGCSに出力するオプションを付ける
 *  --dumpHeapOnOOMと--saveHeapDumpsToGcsPath
* Dataflowワーカーに接続し、ヒープダンプをダウンロードする
* JMX経由で接続する

の三種類が紹介されています。今回はワーカーからのダウンロードを試してみます。


## わかること

* インスタンス毎のJVMのメモリ使用状況
 * 誰がメモリを多く使っている
 * 参照の関係


## 試してみる
以下の手順で行います：

1. パイプラインの起動
2. ヒープダンプの作成
3. ローカルにダウンロード
4. 見てみる

特にオプションや注意点はないので、パイプラインの起動は省略します。

### ヒープダンプの作成
Dataflowワーカーに入ってヒープダンプを作ります。

なお、GCEインスタンスにはDataflowのジョブID・ジョブ名のラベルが付くので、それでインスタンスを特定出来ます（下例）。

```
gcloud compute instances list --filter='labels.dataflow_job_name=${job_name}'
```

SSHトンネルを作ります:

```
gcloud compute ssh --project=$PROJECT --zone=$ZONE \
  $WORKER_NAME --ssh-flag "-L 8081:127.0.0.1:8081"
```

### ローカルにダウンロード
﻿﻿ブラウザでhttp://127.0.0.1:8081/heapzを開きます。
結構時間かかります（n1-standard-4で10分くらい）。


### 見てみる
[VisualVM](https://visualvm.github.io/download.html)とか、好きなツールでダンプを見ませう。
スレッドやオブジェクトの状況が見れます。

![VisualVMサマリー](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/311203/9bc8dd6a-eb83-52fe-114e-8be94c3b61f2.png)

![VisualVMスレッド](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/311203/e1813da7-4a0e-f351-8162-2da4613bb59d.png)

![VisualVMオブジェクト](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/311203/a054d48f-7336-b7b6-a975-b16425c3d9bd.png)

ヒープダンプの見方は先人が色々書いてくれていると思うので、頑張ってください（[Javaパフォーマンス](https://www.oreilly.co.jp/books/9784873117188/)とか）。


## 注意点
今回試したワーカーからのダウンロードに関しては、いくつか注意点がいくつか記載されています：


* ダンプを取ることが、パイプラインのパフォーマンスに影響を与える
* ヒープダンプが一時的にワーカーのディスクに置かれるので、容量に注意
 * 少なくとも30GB+メモリ分は必要らしい