Realtime Compute for Apache Flink:パフォーマンス ホワイトペーパー (Nexmark ベンチマーク)

ステップ 1： テストツールのダウンロード

Nexmark テストツールパッケージ nexmark-flink.tar.gz をダウンロードし、解凍します。

解凍後のディレクトリ構造は以下のとおりです。

nexmark-flink/
├── run_nexmark.sh          # テストエントリスクリプト
├── nexmark_env.sh          # 環境変数設定ファイル (編集が必要)
├── bin/                    # ランタイムスクリプト
├── conf/                   # Flink ジョブ設定
├── lib/                    # JAR ファイル (コンソールにアップロード)
└── queries-vvp/            # Nexmark クエリ SQL ファイル

ステップ 2： Nexmark JAR のアップロード

1. Realtime Compute for Apache Flink コンソール にログインします。

2. 対象のプロジェクトスペースをクリックします。左側のナビゲーションペインで、[ファイル管理] > [リソースのアップロード] をクリックします。

3. nexmark-flink-0.2-SNAPSHOT.jar ファイルを選択してアップロードします。このファイルは、テストツールの nexmark-flink/lib ディレクトリにあります。

4. アップロードが完了したら、ファイル名をクリックして OSS アドレスをコピーします。このアドレスは、後の設定手順で使用します。ファイルパスの形式は、ストレージタイプによって異なります。

   • OSS Bucket ストレージ： oss://<OSS Bucket 名>/artifacts/namespaces/<プロジェクトスペース名>/<ファイル名>

     例： oss://oss-test/artifacts/namespaces/flink-default/nexmark-flink-0.2-SNAPSHOT.jar

   • フルマネージドストレージ： oss://flink-fullymanaged-<ワークスペース ID>/artifacts/namespaces/<プロジェクトスペース名>/<ファイル名>

     例： oss://flink-fullymanaged-e6a123456789/artifacts/namespaces/flink-default/nexmark-flink-0.2-SNAPSHOT.jar

   ワークスペースのストレージタイプを確認するには、Realtime Compute for Apache Flink 管理コンソールにアクセスし、対象のワークスペースを見つけて、[操作] 列の [詳細] をクリックします。

ステップ 3： ランタイムパラメーターの設定

nexmark-flink/nexmark_env.sh ファイルを編集し、以下のパラメーターを設定します。

ステップ 4： テストの実行

1. nexmark-flink ディレクトリで、以下のコマンドを実行します。

   ./run_nexmark.sh

2. テストツールは、OpenAPI を使用して Nexmark ジョブを自動的に作成し、実行します。

3. テストが完了すると、各クエリの実行時間がミリ秒単位で表示されます。例えば、以下の出力ではクエリ q0 の実行時間は 13078 ミリ秒です。

   INFO  com.github.nexmark.flink.vvp.Nexmark - q0 13078
   ============================================================================
   ✓ Benchmark execution completed successfully
   ============================================================================

環境の準備

以下の構成で、ECS 上の EMR を使用して Flink クラスターを作成します。

• EMR バージョン：EMR-5.21.0

• ハードウェア仕様：3 台の ecs.g6a.xlarge インスタンス (4 vCPU / 16 GiB)。マスターノード 1 台とコアノード 2 台を含みます。

• Hadoop および HDFS サービスを有効にします。

• すべてのノード間でパスワードレスログインを設定します。たとえば、秘密鍵ファイル (key.pem など) をマスターノードにアップロードします。次に、マスターノードの ~/.ssh/config ファイルに以下の設定を追加します。IP アドレスとファイルパスは、実際の値に置き換えてください。

  Host 192.168.0.0
      HostName 192.168.0.0
      User root
      IdentityFile /path/to/key.pem
      StrictHostKeyChecking no
  Host 192.168.0.1
      HostName 192.168.0.1
      User root
      IdentityFile /path/to/key.pem
      StrictHostKeyChecking no
  Host 192.168.0.2
      HostName 192.168.0.2
      User root
      IdentityFile /path/to/key.pem
      StrictHostKeyChecking no

  ssh を使用して、ノード間のパスワードレスログインが正しく機能していることを確認してください。「bad permissions」エラーが発生した場合は、chmod 600 /path/to/key.pem を実行して権限を修正してください。

ソフトウェアの準備

1. Flink パッケージ (Apache Flink Downloads) と Nexmark テストパッケージ (nexmark-flink.tgz) をダウンロードします。パッケージをマスターノードにアップロードし、解凍します。

   tar -zxvf flink-1.20.4-bin-scala_2.12.tgz
   tar -zxvf nexmark-flink.tgz
   mv flink-1.20.4 flink
   mv nexmark-flink nexmark

2. nexmark/lib ディレクトリ内の JAR ファイルを flink/lib にコピーします。これらの JAR には、Nexmark データジェネレーターが含まれています。

   cp nexmark/lib/* flink/lib/

3. 環境変数を設定します。~/.bashrc を編集し、以下の設定を追加してから、source ~/.bashrc を実行して変更を反映させます。

   実際の環境に合わせてパスを設定してください。

   export JAVA_HOME=/etc/alternatives/java_sdk_11
   export PATH=$JAVA_HOME/bin:$PATH
   export FLINK_HOME=/mnt/disk1/flink
   export HADOOP_CLASSPATH=$(/opt/apps/HADOOP-COMMON/hadoop-common-current/bin/hadoop classpath)

クラスターの設定と起動

1. Flink ワーカーを設定します。このテストでは 8 つの TaskManager を使用します。マスターノードに 2 つ、各コアノードに 3 つずつデプロイします。

   flink/conf/workers を編集し、IP アドレスを実際の値に置き換えてください。

   192.168.0.0
   192.168.0.0
   192.168.0.1
   192.168.0.1
   192.168.0.1
   192.168.0.2
   192.168.0.2
   192.168.0.2

2. flink/conf/config.yaml を nexmark/conf/config.yaml で置き換え、以下の設定項目を更新します。

   • jobmanager.rpc.address：マスターノードの IP アドレス (192.168.0.0 など)。

   • state.checkpoints.dir：HDFS パス (hdfs:///checkpoints など)。

   • taskmanager.memory.process.size：4 G

3. nexmark/conf/nexmark.yaml を編集し、nexmark.metric.reporter.host をマスターノードの IP アドレスに設定します。

4. flink および nexmark ディレクトリと環境変数設定を、各コアノードに配布します。

   IP アドレスは実際の値に置き換えてください。

   scp -r flink 192.168.0.1:/mnt/disk1/
   scp -r flink 192.168.0.2:/mnt/disk1/
   scp -r nexmark 192.168.0.1:/mnt/disk1/
   scp -r nexmark 192.168.0.2:/mnt/disk1/
   scp ~/.bashrc 192.168.0.1:~/
   scp ~/.bashrc 192.168.0.2:~/

   配布が完了したら、各コアノードで source ~/.bashrc を実行して、環境変数を反映させます。

5. マスターノードで、Flink クラスターを起動します。

   flink/bin/start-cluster.sh

6. Nexmark テスト環境を初期化します。このスクリプトは、各ノードで必要な Metric Reporter を設定します。

   nexmark/bin/setup_cluster.sh

リソース制限の設定

Flink クラスターの起動後、cgroups を使用して各 TaskManager プロセスの CPU 使用率を 75% に制限する必要があります。これにより、リソースの競合によるハートビートタイムアウトが原因で TaskManager が切断されるのを防ぎます。

TaskManager が実行されているすべてのノード (マスターノードを含む) で、以下のコマンドを実行してください。

yum install -y libcgroup libcgroup-tools
cgcreate -t root:root -a root:root -g cpu,memory:mygroup
echo 100000 > /sys/fs/cgroup/cpu/mygroup/cpu.cfs_period_us
echo 300000 > /sys/fs/cgroup/cpu/mygroup/cpu.cfs_quota_us
echo $((12 * 1024 * 1024 * 1024)) > /sys/fs/cgroup/memory/mygroup/memory.limit_in_bytes
jps | grep TaskManagerRunner | awk '{print $1}' | xargs cgclassify -g cpu,memory:mygroup

ここで、cpu.cfs_quota_us / cpu.cfs_period_us = 300000 / 100000 = 3 です。これは、cgroup が最大 3 CPU コア (4 vCPU の 75%) を使用できることを意味します。

Nexmark の実行

マスターノードで以下のコマンドを実行します。完了するまで待機し、結果を確認してください。

nexmark/bin/run_query.sh q0,q1,q2,q3,q4,q5,q7,q8,q9,q10,q11,q12,q14,q15,q16,q17,q18,q19,q20,q21,q22