Flink を設定する - Amazon EMR

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

Flink を設定する

設定ファイルを使用して Flink を設定します。たとえば、Flink の主となる設定ファイルは flink-conf.yaml と呼ばれます。このファイルは、Amazon EMR 設定 API を使用して設定することができます。

AWS CLI を使用して Flink のタスクのスロット数を設定するには

  1. 次のコンテンツを含む configurations.json ファイルを作成します。

    [ { "Classification": "flink-conf", "Properties": { "taskmanager.numberOfTaskSlots":"2" } } ]
  2. 次に、次の設定でクラスターを作成します。

    aws emr create-cluster --release-label emr-5.36.0 \ --applications Name=Flink \ --configurations file://./configurations.json \ --region us-east-1 \ --log-uri s3://myLogUri \ --instance-type m5.xlarge \ --instance-count 2 \ --service-role EMR_DefaultRole \ --ec2-attributes KeyName=YourKeyName,InstanceProfile=EMR_EC2_DefaultRole
注記

Flink API を使用して設定を変更することもできます。詳細については、Flink のドキュメントで「Concepts (概念)」を確認してください。

Amazon EMR バージョン 5.21.0 以降では、実行中のクラスター内のインスタンスグループごとに、クラスター設定を上書きして追加の設定分類を指定できます。これを行うには、Amazon EMR コンソール、AWS Command Line Interface (AWS CLI)、または AWS SDK を使用します。詳細については、「実行中のクラスター内のインスタンスグループの設定を指定する」を参照してください。

アプリケーションの所有者は、どのリソースが Flink 内のタスクに割り当てられるべきかを最もよく知っています。このドキュメントでは、例として、アプリケーション用に使っているスレーブインスタンスと同じ数のタスクを使用します。一般的には、並列処理の最初の段階にこれをすすめていますが、タスクスロットを使用して並列処理の粒度を高めることも可能です。こうすると、通常、インスタンスごとの仮想コア数を超過することはありません。Flink のアーキテクチャの詳細については、Flink ドキュメントで「Concepts (概念)」をご確認ください。

現在、Amazon EMR 設定 API で設定可能なファイルは次のとおりです。

  • flink-conf.yaml

  • log4j.properties

  • flink-log4j-session

  • log4j-cli.properties

- JobManager of Flink は、複数のマスターノードを持つ EMR クラスターのマスターノードフェイルオーバープロセス中も引き続き使用できます。Amazon EMR バージョン 5.28.0 以降、 JobManager 高可用性も自動的に有効になります。手動設定は必要ありません。

Amazon EMR バージョン 5.27.0 以前では、 JobManager 単一障害点です。のとき JobManager 失敗すると、すべてのジョブの状態が失われ、実行中のジョブは再開されません。有効にできます JobManager アプリケーションの試行回数、チェックポイント、有効化の設定による高可用性 ZooKeeper Flink の State ストレージとして、以下の例にあるように、Flink の State ストレージとして使用します。

[ { "Classification": "yarn-site", "Properties": { "yarn.resourcemanager.am.max-attempts": "10" } }, { "Classification": "flink-conf", "Properties": { "yarn.application-attempts": "10", "high-availability": "zookeeper", "high-availability.zookeeper.quorum": "%{hiera('hadoop::zk')}", "high-availability.storageDir": "hdfs:///user/flink/recovery", "high-availability.zookeeper.path.root": "/flink" } } ]

YARN の最大アプリケーションマスター試行回数と Flink のアプリケーション試行回数の両方を設定する必要があります。詳細については、「YARN クラスターの高可用性の設定」を参照してください。また、Flink のチェックポインティングを再起動するように設定することもできます。 JobManager 以前に完了したチェックポイントから実行中のジョブを復元します。詳細については、「Flink のチェックポイント」を参照してください。

Flink 1.11.x を使用する Amazon EMR バージョンでは、両方の合計メモリプロセスサイズを設定する必要があります JobManager(jobmanager.memory.process.size) と TaskManager(taskmanager.memory.process.size) にflink-conf.yaml。これらの値を設定するには、設定 API を使用してクラスターを設定するか、SSH を介してこれらのフィールドのコメントを手動で解除します。Flink は以下のデフォルト値を提供します。

  • jobmanager.memory.process.size: 1600m

  • taskmanager.memory.process.size: 1728m

JVM メタスペースとオーバーヘッドを除外するには、Flink の合計メモリサイズ (taskmanager.memory.flink.size) の代わりにtaskmanager.memory.process.size。のデフォルト値taskmanager.memory.process.size1280mです。taskmanager.memory.process.sizetaskmanager.memory.process.size の両方を設定することはお勧めしません。

Flink 1.12.0 以降を使用するすべての Amazon EMR バージョンでは、Flink のオープンソースにリストされているデフォルト値が Amazon EMR のデフォルト値として設定されているため、自分で設定する必要はありません。

Flink アプリケーションコンテナは、.out ファイル、.log ファイル、.err ファイルの 3 種類のログファイルを作成して書き込みます。.err ファイルのみが圧縮されてファイルシステムから削除され、.log および .out ログファイルはファイルシステムに残ります。これらの出力ファイルを管理しやすく、クラスターを安定させるために、ログローテーションを設定できます。log4j.properties最大ファイル数を設定し、サイズを制限します。

Amazon EMR バージョン 5.30.0 以降

Amazon EMR 5.30.0 以降、Flink は設定分類名の log4j2 ロギングフレームワークを使用します。flink-log4j.次の設定例は、log4j2 形式を示しています。

[ { "Classification": "flink-log4j", "Properties": { "appender.rolling.name": "RollingFileAppender", "appender.rolling.type":"RollingFile", "appender.rolling.append" : "false", "appender.rolling.fileName" : "${sys:log.file}", "appender.rolling.filePattern" : "${sys:log.file}.%i", "appender.rolling.layout.type" : "PatternLayout", "appender.rolling.layout.pattern" : "%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n", "appender.rolling.policies.type" : "Policies", "appender.rolling.policies.size.type" : "SizeBasedTriggeringPolicy", "appender.rolling.policies.size.size" : "100MB", "appender.rolling.strategy.type" : "DefaultRolloverStrategy", "appender.rolling.strategy.max" : "10" }, } ]

Amazon EMR バージョン 5.29.0 以前

Amazon EMR バージョン 5.29.0 以前では、Flink は log4j ロギングフレームワークを使用しています。次の設定例は、log4j 形式を示しています。

[ { "Classification": "flink-log4j", "Properties": { "log4j.appender.file": "org.apache.log4j.RollingFileAppender", "log4j.appender.file.append":"true", # keep up to 4 files and each file size is limited to 100MB "log4j.appender.file.MaxFileSize":"100MB", "log4j.appender.file.MaxBackupIndex":4, "log4j.appender.file.layout":"org.apache.log4j.PatternLayout", "log4j.appender.file.layout.ConversionPattern":"%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n" }, } ]