ハイブ・アンド・Glue 設定ファイル複数のプライマリノードメモリプロセスサイズログ出力ファイルのサイズ Java 11

Amazon EMR でのフリンクの設定

Hive メタストアと Glue カタログによる Flink の設定

Amazon EMR リリース 6.9.0 以降は、Hive への Apache Flink コネクタを備えた Hive メタストアと AWS Glue カタログの両方をサポートしています。このセクションでは、Flink で AWSGlue カタログと Hive メタストアを設定するために必要な手順の概要を説明します。

トピック

Hive メタストアを使用する
AWSGlue データカタログを使用する

Hive メタストアを使用する

リリース 6.9.0 以降と、少なくとも 2 つのアプリケーション (Hive と Flink) で EMR クラスターを作成します。

スクリプトランナーを使用して、以下のスクリプトをステップ関数として実行します。

hive-metastore-setup.sh


sudo cp /usr/lib/hive/lib/antlr-runtime-3.5.2.jar /usr/lib/flink/lib 
sudo cp /usr/lib/hive/lib/hive-exec-3.1.3*.jar /lib/flink/lib 
sudo cp /usr/lib/hive/lib/libfb303-0.9.3.jar /lib/flink/lib 
sudo cp /usr/lib/flink/opt/flink-connector-hive_2.12-1.15.2.jar /lib/flink/lib
sudo chmod 755 /usr/lib/flink/lib/antlr-runtime-3.5.2.jar 
sudo chmod 755 /usr/lib/flink/lib/hive-exec-3.1.3*.jar 
sudo chmod 755 /usr/lib/flink/lib/libfb303-0.9.3.jar
sudo chmod 755 /usr/lib/flink/lib/flink-connector-hive_2.12-1.15.2.jar

AWSGlue データカタログを使用する

リリース 6.9.0 以降と、少なくとも 2 つのアプリケーション (Hive と Flink) で EMR クラスターを作成します。
AWSGlue データカタログ設定で「Hive テーブルメタデータに使用」を選択して、クラスター内のデータカタログを有効にします。

script runner を使用して次のスクリプトをステップ関数として実行します。Amazon EMR クラスターでコマンドとスクリプトを実行します。

glue-catalog-setup.sh


sudo cp /usr/lib/hive/auxlib/aws-glue-datacatalog-hive3-client.jar /usr/lib/flink/lib 
sudo cp /usr/lib/hive/lib/antlr-runtime-3.5.2.jar /usr/lib/flink/lib 
sudo cp /usr/lib/hive/lib/hive-exec-3.1.3*.jar /lib/flink/lib 
sudo cp /usr/lib/hive/lib/libfb303-0.9.3.jar /lib/flink/lib 
sudo cp /usr/lib/flink/opt/flink-connector-hive_2.12-1.15.2.jar /lib/flink/lib
sudo chmod 755 /usr/lib/flink/lib/aws-glue-datacatalog-hive3-client.jar 
sudo chmod 755 /usr/lib/flink/lib/antlr-runtime-3.5.2.jar 
sudo chmod 755 /usr/lib/flink/lib/hive-exec-3.1.3*.jar 
sudo chmod 755 /usr/lib/flink/lib/libfb303-0.9.3.jar
sudo chmod 755 /usr/lib/flink/lib/flink-connector-hive_2.12-1.15.2.jar

設定ファイルを使用して Flink を設定します。

Amazon EMR 設定 API を使用して、設定ファイルを使用して Flink を設定できます。API 内で設定できるファイルは次のとおりです。

flink-conf.yaml
log4j.properties
flink-log4j-session
log4j-cli.properties

Flink のメイン設定ファイルはです。flink-conf.yaml

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

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


[
    {
      "Classification": "flink-conf",
      "Properties": {
        "taskmanager.numberOfTaskSlots":"2"
      }
    }
]

次に、次の設定でクラスターを作成します。


aws emr create-cluster --release-label emr-5.36.1 \
--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_V2 \ 
--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 を使用します。詳細については、「実行中のクラスター内のインスタンスグループの設定を指定する」を参照してください。

Parallelism オプション

アプリケーションのオーナーは、Flink 内のタスクに割り当てるリソースを最もよく知っています。このドキュメントの例では、アプリケーションに使用するタスクインスタンスと同じ数のタスクを使用してください。通常、並列処理の初期レベルではこれをお勧めしますが、タスクスロットを使用して並列処理を細かくすることもできます。タスクスロットは、通常、インスタンスあたりの仮想コア数を超えないようにしてください。Flink アーキテクチャの詳細については、Flink ドキュメントの「概念」を参照してください。

複数のプライマリノードがある EMR クラスターでの Flink の設定

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

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


[
  {
    "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のデフォルト値は 1280 m です。taskmanager.memory.process.sizetaskmanager.memory.process.size と taskmanager.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 ロギングフレームワークを使用します。次の設定例は log4j2 flink-log4j. 形式を示しています。


[
  {
    "Classification": "flink-log4j",
    "Properties": {
      "rootLogger.appenderRef.rolling.ref": "RollingFileAppender",
      "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"
    },
  }
]

Flink を Java 11 で実行するように設定します。

Amazon EMR リリース 6.12.0 以降では、Flink の Java 11 ランタイムサポートが提供されています。以下のセクションでは、Flink に Java 11 ランタイムサポートを提供するようにクラスターを設定する方法について説明します。

クラスターを作成するときに Flink for Java 11 を設定します。

次の手順を使用して、Flink と Java 11 ランタイムで EMR クラスターを作成します。Java 11 ランタイムサポートを追加する設定ファイルはです。flink-conf.yaml

New console

Flink と Java 11 ランタイムを含むクラスターを新しいコンソールで作成するには

にサインインしAWS Management Console、https://console.aws.amazon.com/emr にある Amazon EMR コンソールを開きます。
ナビゲーションペインの [EMR on EC2] の下の [クラスター] を選択し、[クラスターの作成] を選択します。
Amazon EMR リリース 6.12.0 以降を選択し、Flink アプリケーションのインストールを選択します。クラスターにインストールする他のアプリケーションを選択します。

クラスターの設定を続行します。オプションの「ソフトウェア設定」セクションで、デフォルトの 「設定を入力」オプションを使用して、次の設定を入力します。


[
    {
      "Classification": "flink-conf",
      "Properties": {
        "containerized.taskmanager.env.JAVA_HOME":"/usr/lib/jvm/jre-11",
        "containerized.master.env.JAVA_HOME":"/usr/lib/jvm/jre-11",
        "env.java.home":"/usr/lib/jvm/jre-11"
      }
    }
]

続けてクラスターの設定と起動を行います。

AWS CLI

CLI から Flink と Java 11 ランタイムを使用してクラスターを作成するには

Flink が Java 11 configurations.json を使用するように設定する設定ファイルを作成します。


[
    {
      "Classification": "flink-conf",
      "Properties": {
        "containerized.taskmanager.env.JAVA_HOME":"/usr/lib/jvm/jre-11",
        "containerized.master.env.JAVA_HOME":"/usr/lib/jvm/jre-11",
        "env.java.home":"/usr/lib/jvm/jre-11"
      }
    }
]

からAWS CLI、Amazon EMR リリース 6.12.0 以降で新しい EMR クラスターを作成し、次の例に示すように Flink アプリケーションをインストールします。


aws emr create-cluster --release-label emr-6.12.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_V2 \ 
--ec2-attributes KeyName=YourKeyName,InstanceProfile=EMR_EC2_DefaultRole

実行中のクラスターに Flink for Java 11 を設定します。

次の手順を使用して、実行中の EMR クラスターを Flink と Java 11 ランタイムで更新します。Java 11 ランタイムサポートを追加する設定ファイルはです。flink-conf.yaml

New console

新しいコンソールで、実行中のクラスターを Flink と Java 11 ランタイムで更新するには

にサインインしAWS Management Console、https://console.aws.amazon.com/emr にある Amazon EMR コンソールを開きます。
ナビゲーションペインの [EMR on EC2] の下の [Clusters] を選択し、更新するクラスターを選択します。

注記
クラスターが Java 11 をサポートするには、Amazon EMR リリース 6.12.0 以降を使用する必要があります。
[設定] タブを選択します。
「インスタンスグループ設定」セクションで、更新する実行中のインスタンスグループを選択し、リストアクションメニューから「再設定」を選択します。

[属性の編集] オプションを使用してインスタンスグループを次のように再設定します。設定を変更するたびに [新しい設定を追加] を選択します。

分類	プロパティ	値
`flink-conf`	`containerized.taskmanager.env.JAVA_HOME`	`/usr/lib/jvm/jre-11`
`flink-conf`	`containerized.master.env.JAVA_HOME`	`/usr/lib/jvm/jre-11`
`flink-conf`	`env.java.home`	`/usr/lib/jvm/jre-11`

[変更を保存] を選択して設定を追加します。

AWS CLI

CLI から Flink と Java 11 ランタイムを使用するように実行中のクラスターを更新するには

modify-instance-groups コマンドを使用して、実行中のクラスター内のインスタンスグループに新しい設定を指定します。

まず、Flink が Java 11 configurations.json を使用するように設定する設定ファイルを作成します。次の例では、ig-1xxxxx9 を再構成するインスタンスグループの ID に置き換えます。コマンドを実行するのと同じディレクトリにファイルを保存します。modify-instance-groups


[
   {
      "InstanceGroupId":"ig-1xxxxxxx9",
      "Configurations":[
         {
            "Classification":"flink-conf",
            "Properties":{
              "containerized.taskmanager.env.JAVA_HOME":"/usr/lib/jvm/jre-11",
              "containerized.master.env.JAVA_HOME":"/usr/lib/jvm/jre-11",
              "env.java.home":"/usr/lib/jvm/jre-11"
            },
            "Configurations":[]
         }
      ]
   }
]

からAWS CLI、以下のコマンドを実行します。再構成するインスタンスグループの ID を置き換えます。
```
aws emr modify-instance-groups --cluster-id j-2AL4XXXXXX5T9 \
--instance-groups file://configurations.json
```

実行中のクラスターで Flink の Java ランタイムを確認します。

実行中のクラスターの Java ランタイムを確認するには、「SSH によるプライマリノードへのConnect」の説明に従って SSH でプライマリノードにログインします。次に、以下のコマンドを実行します。


ps -ef | grep flink

ps-efオプションを指定したコマンドは、システム上で実行中のすべてのプロセスを一覧表示します。grepでその出力をフィルタリングして、flink文字列に関する記述を検索できます。Java ランタイム環境 (JRE) の値の出力を確認します。jre-XX次の出力では、実行時に Flink に Java 11 jre-11 が起動されていることが示されています。

flink    19130     1  0 09:17 ?        00:00:15 /usr/lib/jvm/jre-11/bin/java -Djava.io.tmpdir=/mnt/tmp -Dlog.file=/usr/lib/flink/log/flink-flink-historyserver-0-ip-172-31-32-127.log -Dlog4j.configuration=file:/usr/lib/flink/conf/log4j.properties -Dlog4j.configurationFile=file:/usr/lib/flink/conf/log4j.properties -Dlogback.configurationFile=file:/usr/lib/flink/conf/logback.xml -classpath /usr/lib/flink/lib/flink-cep-1.17.0.jar:/usr/lib/flink/lib/flink-connector-files-1.17.0.jar:/usr/lib/flink/lib/flink-csv-1.17.0.jar:/usr/lib/flink/lib/flink-json-1.17.0.jar:/usr/lib/flink/lib/flink-scala_2.12-1.17.0.jar:/usr/lib/flink/lib/flink-table-api-java-uber-1.17.0.jar:/usr/lib/flink/lib/flink-table-api-scala-bridge_2.12-1.17.0.

または、SSH を使用してプライマリノードにログインし、コマンドで Flink YARN セッションを開始します。flink-yarn-session -d次の例では、出力に Flink 用の Java 仮想マシン (JVM) が表示されます。java-11-amazon-corretto

2023-05-29 10:38:14,129 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: containerized.master.env.JAVA_HOME, /usr/lib/jvm/java-11-amazon-corretto.x86_64

ブラウザで JavaScript が無効になっているか、使用できません。

AWS ドキュメントを使用するには、JavaScript を有効にする必要があります。手順については、使用するブラウザのヘルプページを参照してください。

ドキュメントの表記規則

Flink を使用してクラスターを作成する

Flink ジョブ