E-MapReduce:よくある質問

クラスターログの表示

JAR パッケージの競合の解決

外部マシンからのジョブの送信

外部マシンからのクラスターホスト名の解決

Flink ジョブのステータスの確認

• E-MapReduce (EMR) コンソールを使用します。

  EMR は Knox をサポートしており、インターネット経由で YARN や Flink などのサービスの Web UI にアクセスできます。YARN を通じて Flink Web UI にアクセスできます。詳細については、「Web UI でジョブステータスを表示」をご参照ください。

• SSH トンネルを使用します。詳細については、「SSH トンネルを作成してオープンソースコンポーネントの Web UI にアクセスする」をご参照ください。

• YARN REST API に直接アクセスします。

  curl --compressed -v  -H "Accept: application/json" -X GET "http://master-1-1:8088/ws/v1/cluster/apps?states=RUNNING&queue=default&user.name=***"

  説明

  YARN REST API にアクセスするには、セキュリティグループでポート 8443 と 8088 へのアクセスが許可されていることを確認してください。または、DataFlow クラスターとクライアントノードが同じ Virtual Private Cloud (VPC) 内にあることを確認してください。

Flink ジョブログへのアクセス

• 実行中のジョブの場合、Flink Web UI を通じてそのログにアクセスできます。

• 完了したジョブの場合、Flink HistoryServer でその統計情報を表示するか、yarn logs -applicationId application_xxxx_yyyy コマンドを実行してそのログにアクセスできます。完了したジョブのログは、デフォルトで HDFS クラスターの hdfs:///tmp/logs/$USERNAME/logs/ ディレクトリに保存されます。

Flink HistoryServer へのアクセス

商用コネクタの使用

DataFlow クラスターは、Hologres、SLS、MaxCompute、DataHub、Elasticsearch、ClickHouse など、多くの商用コネクタを提供します。Flink ジョブでは、オープンソースのコネクタに加えて、これらの商用コネクタを使用できます。次の例は、付属の Hologres コネクタの使用方法を示しています。

• ジョブ開発

  1. DataFlow クラスターから商用コネクタの JAR パッケージをダウンロードします (場所は /opt/apps/FLINK/flink-current/opt/connectors ディレクトリ)。次に、次のコマンドを実行して、ローカルの Maven 環境にコネクタをインストールしてください。

     mvn install:install-file -Dfile=/path/to/ververica-connector-hologres-1.13-vvr-4.0.7.jar -DgroupId=com.alibaba.ververica -DartifactId=ververica-connector-hologres -Dversion=1.13-vvr-4.0.7 -Dpackaging=jar

  2. プロジェクトの pom.xml ファイルに次の依存関係を追加してください。

     <dependency>
         <groupId>com.alibaba.ververica</groupId>
         <artifactId>ververica-connector-hologres</artifactId>
         <version>1.13-vvr-4.0.7</version>
         <scope>provided</scope>
     </dependency>

• ジョブの実行

  • 方法 1：

    1. Hologres コネクタを別のディレクトリにコピーしてください。

       hdfs mkdir hdfs:///flink-current/opt/connectors/hologres/
       hdfs cp hdfs:///flink-current/opt/connectors/ververica-connector-hologres-1.13-vvr-4.0.7.jar  hdfs:///flink-current/opt/connectors/hologres/ververica-connector-hologres-1.13-vvr-4.0.7.jar

    2. ジョブを送信する際に、コマンドに次のパラメーターを追加してください。

       -D yarn.provided.lib.dirs=hdfs:///flink-current/opt/connectors/hologres/

  • 方法 2：

    1. Hologres コネクタを、ジョブ送信クライアントの /opt/apps/FLINK/flink-current/opt/connectors/ververica-connector-hologres-1.13-vvr-4.0.7.jar ディレクトリにコピーします。このディレクトリ構造は、DataFlow クラスター内のものと一致している必要があります。

    2. ジョブを送信する際に、コマンドに次のパラメーターを追加してください。

       -C file:///opt/apps/FLINK/flink-current/opt/connectors/ververica-connector-hologres-1.13-vvr-4.0.7.jar

  • 方法 3：Hologres コネクタをジョブの JAR パッケージにパッケージ化します。

GeminiStateBackend の使用

DataFlow クラスターは、オープンソース版の 3～5 倍のパフォーマンスを提供するエンタープライズ級の GeminiStateBackend を提供します。DataFlow クラスターは、デフォルトで GeminiStateBackend を使用します。GeminiStateBackend の高度な設定に関する詳細については、「エンタープライズ級のステートバックエンド設定」をご参照ください。

オープンソースのステートバックエンドの使用

flink run-application -t yarn-application -D state.backend=rocksdb  /opt/apps/FLINK/flink-current/examples/streaming/TopSpeedWindowing.jar

クライアントログの表示

EMR クラスター環境では、FLINK_LOG_DIR 環境変数が Flink クライアントログの保存場所を指定します。そのデフォルト値は /var/log/taihao-apps/flink です (3.43.0 より前のバージョンでは、デフォルトは /mnt/disk1/log/flink でした)。SQL Client ログなどの完全なクライアントログを表示する必要がある場合は、このディレクトリで対応するファイルを見つけることができます。

ジョブパラメーターが有効にならない問題

コマンドラインから Flink ジョブを実行する場合、ジョブパラメーターを Flink ジョブの JAR パッケージの後に配置してください。例：flink run -d -t yarn-per-job test.jar arg1 arg2。

「Multiple factories...」エラーの解決

• 原因

  このエラーは、クラスパスにコネクタの複数の実装が含まれていることを示します。これは通常、ジョブの JAR パッケージにコネクタの依存関係を追加し、同時に同じコネクタの依存関係を $FLINK_HOME/lib ディレクトリに手動で配置した場合に発生し、依存関係の競合を引き起こします。

• 解決策

  解決策は、重複する依存関係を削除することです。詳細なトラブルシューティング手順については、「ジョブの JAR パッケージがクラスターの Flink JAR パッケージと競合した場合はどうすればよいですか？」をご参照ください。

JobManager HA の有効化

high-availability: zookeeper
high-availability.zookeeper.quorum: 192.168.**.**:2181,192.168.**.**:2181,192.168.**.**:2181
high-availability.zookeeper.path.root: /flink
high-availability.storageDir: hdfs:///flink/recovery

Flink ジョブのメトリクスの表示

1. EMR コンソールで、ターゲットクラスターの Monitoring ページに移動して、Metric Monitoring をクリックしてください。

2. [Dashboard] ドロップダウンリストから [FLINK] を選択してください。

3. 表示したいジョブのアプリケーション ID とジョブ ID を選択してください。すると、ジョブのモニタリングメトリクスが表示されます。

   説明

   • アプリケーション ID とジョブ ID のオプションは、クラスターで Flink ジョブが実行されている場合にのみ利用可能です。

   • sourceIdleTime のような一部のメトリクスは、対応するソースとシンクが設定されている場合にのみ生成されます。

コネクタの問題のトラブルシューティング

アップストリームおよびダウンストリームストレージに関する一般的な質問については、「コネクタ」をご参照ください。

パスワードなしの OSS アクセスでのエラーの解決

特定のエラーメッセージに基づいて問題を処理してください：

• エラーメッセージ：java.lang.UnsupportedOperationException: Recoverable writers on Hadoop are only supported for HDFS

  • 原因：DataFlow クラスターは、組み込みの JindoSDK を使用して、OSS へのパスワードなしのアクセスと StreamingFileSink などの API をサポートします。コミュニティドキュメントに記載されているような追加の設定を行う必要はありません。これを行うと、依存関係の競合が発生し、このエラーが発生する可能性があります。

  • 解決策：クラスターのジョブ送信マシンで、$FLINK_HOME/plugins ディレクトリに oss-fs-hadoop ディレクトリがあるかどうかを確認します。存在する場合は、そのディレクトリを削除してジョブを再送信してください。

• エラーメッセージ：Could not find a file system implementation for scheme 'oss'. The scheme is directly supported by Flink through the following plugin: flink-oss-fs-hadoop. ....

  • 原因：バージョン 3.40 以前の EMR クラスターでは、master-1-1 以外のマスターサーバーグループのマシンに Jindo 関連の JAR パッケージがない場合があります。

  • 解決策：

    • EMR 3.40 以前の場合：ジョブ送信マシンの $FLINK_HOME/lib ディレクトリに jindo-flink-4.0.0-full.jar などの Jindo 関連の JAR パッケージが存在するかどうかを確認します。存在しない場合は、次のコマンドを実行して必要な JAR パッケージを $FLINK_HOME/lib ディレクトリにコピーして、ジョブを再送信してください。

      cp /opt/apps/extra-jars/flink/jindo-flink-*-full.jar $FLINK_HOME/lib

    • EMR 3.40 より後のバージョンの場合：

      • Flink on YARN モードの場合：新しいバージョンでは OSS サポートのメカニズムが最適化されています。OSS の読み書きを行うジョブは、$FLINK_HOME/lib ディレクトリに Jindo 関連の JAR パッケージが存在しなくても正常に実行できます。

      • その他のデプロイメントモードの場合：ジョブ送信マシンの $FLINK_HOME/lib ディレクトリに jindo-flink-4.0.0-full.jar などの Jindo 関連の JAR パッケージが存在するかどうかを確認します。存在しない場合は、次のコマンドを実行して $FLINK_HOME/lib ディレクトリにコピーして、ジョブを再送信してください。

        cp /opt/apps/extra-jars/flink/jindo-flink-*-full.jar $FLINK_HOME/lib

「TaskManager heartbeat timed out」エラーの解決

• 原因

  直接的な原因は TaskManager のハートビートのタイムアウトです。TaskManager のログで特定のエラーメッセージを確認することで、正確な原因を特定できます。その他の潜在的な原因には、TaskManager のヒープメモリ不足や、ジョブコードのメモリリークによるメモリ不足 (OOM) エラーなどがあります。詳細については、「「java.lang.OutOfMemoryError: GC overhead limit exceeded」エラーを解決するにはどうすればよいですか？」をご参照ください。

• 解決策

  このエラーが発生した場合は、メモリ割り当てを増やすか、ジョブのメモリ使用量を分析して問題をさらに診断してください。

「GC overhead limit exceeded」エラーの解決

• 原因

  このエラーは、ジョブに割り当てられたメモリが不足しているために、ガベージコレクター (GC) に時間がかかりすぎていることを示します。一般的な原因には、コード (UDF など) のメモリリークや、ジョブの要件に対して設定されたメモリが不足していることなどがあります。

• 解決策

  • ジョブを再実行する前に、-D フラグを使用して次の JVM パラメーターを指定して、OutOfMemoryError が発生したときにヒープダンプが保存されるようにしてください：-D env.java.opts="-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/dump.hprof"。

  • flink-conf.yaml ファイルにパラメーター env.java.opts: -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/dump.hprof を追加して、OutOfMemoryError 時にヒープダンプを構成してください。

  エラーが再度発生した後、MAT や jvisualvm などのツールを使用して HeapDumpPath で指定されたヒープダンプファイルを分析し、根本原因を特定できます。

単一オペレーターのジョブで「Records Received」がゼロになる問題

これは正常な動作です。Flink の Records Received メトリクスは、異なるオペレーター間のデータ通信を表します。ジョブが単一のオペレーターに最適化されると、このメトリクスは常に 0 になります。

Flink ジョブのフレームグラフの有効化

フレームグラフは、プロセス内のさまざまなメソッドの CPU 消費量を可視化し、パフォーマンスのボトルネックを特定して解決するのに役立ちます。Flink はバージョン 1.13 以降でフレームグラフをサポートしていますが、本番ジョブへの影響を避けるために、この機能はデフォルトで無効になっています。フレームグラフを使用してジョブのパフォーマンスを分析する必要がある場合は、EMR コンソールの Flink サービスの Configure タブに移動してください。[flink-conf.yaml] ファイルで、パラメーター rest.flamegraph.enabled を持つ新しい設定項目を追加して、その値を true に設定してください。設定項目の追加手順については、「設定項目の管理」をご参照ください。

フレームグラフの詳細については、「フレームグラフ」をご参照ください。

「NoSuchFieldError: DEPLOYMENT_MODE」エラーの解決

• 原因

  ジョブの JAR パッケージに、クラスター内の Flink バージョンと互換性のない flink-core 依存関係が直接的または間接的に含まれており、依存関係の競合を引き起こしています。

• 解決策

  pom.xml ファイルに次の設定を追加して、flink-core 依存関係の scope を provided に設定してください。これにより問題が解決します。

  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <!-- 実際の Flink バージョンに変更してください -->
    <version>1.16.1</version>
    <scope>provided</scope>
  </dependency>

  説明

  version をご自身の Flink バージョンに変更する必要があります。

  この依存関係のソースをさらに特定するには、「ジョブの JAR パッケージがクラスターの Flink JAR パッケージと競合した場合はどうすればよいですか？」をご参照ください。