Mengkonfigurasi Flink di Amazon EMR - Amazon EMR
Sarang dan GlueBerkas ConfigBeberapa node utamaUkuran proses memoriUkuran file keluaran logJava 11

Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Mengkonfigurasi Flink di Amazon EMR

Amazon EMR merilis 6.9.0 dan lebih tinggi mendukung Hive Metastore dan Glue Catalog dengan konektor Apache AWS Flink ke Hive. Bagian ini menguraikan langkah-langkah yang diperlukan untuk mengkonfigurasi AWS Glue Catalog dan Hive Metastore dengan Flink.

  1. Buat cluster EMR dengan rilis 6.9.0 atau lebih tinggi dan setidaknya dua aplikasi: Hive dan Flink.

  2. Gunakan script runner untuk menjalankan script berikut sebagai fungsi langkah:

    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
    Form to add a Custom JAR step with fields for step type, name, JAR location, arguments, and failure action.

  1. Buat cluster EMR dengan rilis 6.9.0 atau lebih tinggi dan setidaknya dua aplikasi: Hive dan Flink.

  2. Pilih Gunakan untuk metadata tabel Hive di pengaturan AWS Glue Data Catalog untuk mengaktifkan Data Catalog di cluster.

  3. Gunakan script runner untuk menjalankan skrip berikut sebagai fungsi langkah: Jalankan perintah dan skrip pada klaster EMR Amazon:

    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
    Form to add a Custom JAR step with fields for step type, name, JAR location, arguments, and failure action.

Anda dapat menggunakan API konfigurasi EMR Amazon untuk mengonfigurasi Flink dengan file konfigurasi. File yang dapat dikonfigurasi dalam API adalah:

  • flink-conf.yaml

  • log4j.properties

  • flink-log4j-session

  • log4j-cli.properties

File konfigurasi utama untuk Flink adalahflink-conf.yaml.

Untuk mengkonfigurasi jumlah slot tugas yang digunakan untuk Flink dari AWS CLI

  1. Buat file, configurations.json, dengan konten berikut:

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

  2. Berikutnya, buat sebuah klaster dengan konfigurasi berikut:

    aws emr create-cluster --release-label emr-7.10.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
catatan

Anda juga dapat mengubah beberapa konfigurasi dengan Flink API. Untuk informasi lebih lanjut, lihat Konsep di dokumentasi Flink.

Dengan Amazon EMR versi 5.21.0 dan yang lebih baru, Anda dapat mengganti konfigurasi klaster dan menentukan klasifikasi konfigurasi tambahan untuk setiap grup instans dalam klaster berjalan. Anda melakukannya dengan menggunakan konsol EMR Amazon, AWS Command Line Interface (AWS CLI), atau SDK AWS . Untuk informasi selengkapnya, lihat Menyediakan Konfigurasi untuk Grup Instans dalam Klaster Berjalan.

Sebagai pemilik aplikasi Anda, Anda paling tahu sumber daya apa yang harus ditetapkan ke tugas dalam Flink. Untuk contoh dalam dokumentasi ini, gunakan jumlah tugas yang sama dengan instance tugas yang Anda gunakan untuk aplikasi. Kami biasanya merekomendasikan ini untuk tingkat awal paralelisme, tetapi Anda juga dapat meningkatkan granularitas paralelisme dengan slot tugas, yang umumnya tidak boleh melebihi jumlah inti virtual per instance. Untuk informasi selengkapnya tentang arsitektur Flink, lihat Konsep dalam dokumentasi Flink.

Flink tetap tersedia selama proses failover node primer di cluster EMR Amazon dengan beberapa node primer. JobManager Dimulai dengan Amazon EMR 5.28.0, ketersediaan JobManager tinggi juga diaktifkan secara otomatis. Tidak ada konfigurasi manual yang diperlukan.

Dengan Amazon EMR versi 5.27.0 atau sebelumnya, ini JobManager adalah satu titik kegagalan. Ketika JobManager gagal, ia kehilangan semua status pekerjaan dan tidak akan melanjutkan pekerjaan yang sedang berjalan. Anda dapat mengaktifkan ketersediaan JobManager tinggi dengan mengonfigurasi jumlah upaya aplikasi, pos pemeriksaan, dan mengaktifkan ZooKeeper sebagai penyimpanan status untuk Flink, seperti yang ditunjukkan contoh berikut:

[
  {
    "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"
    }
  }
]

Anda harus mengonfigurasi kedua upaya master aplikasi maksimum untuk YARN dan upaya aplikasi untuk Flink. Untuk informasi lebih lanjut, lihat Konfigurasi ketersediaan tinggi klaster YARN. Anda mungkin juga ingin mengonfigurasi pos pemeriksaan Flink untuk membuat restart JobManager memulihkan pekerjaan yang berjalan dari pos pemeriksaan yang telah diselesaikan sebelumnya. Untuk informasi lebih lanjut, lihat Titik pemeriksaan Flink.

Untuk versi EMR Amazon yang menggunakan Flink 1.11.x, Anda harus mengonfigurasi ukuran proses memori total untuk () dan JobManager (jobmanager.memory.process.size) di. TaskManager taskmanager.memory.process.size flink-conf.yaml Anda dapat mengatur nilai-nilai ini dengan mengonfigurasi klaster dengan API konfigurasi atau secara manual tidak berkomentar pada bidang ini melalui SSH. Flink memberikan nilai default berikut.

  • jobmanager.memory.process.size: 1600m

  • taskmanager.memory.process.size: 1728m

Untuk mengecualikan metaspace dan overhead JVM, gunakan total ukuran memori Flink () alih-alih. taskmanager.memory.flink.size taskmanager.memory.process.size Nilai default untuk taskmanager.memory.process.size adalah 1280m. Ini tidak direkomendasikan untuk mengatur taskmanager.memory.process.size dan taskmanager.memory.process.size.

Semua versi EMR Amazon yang menggunakan Flink 1.12.0 dan yang lebih baru memiliki nilai default yang tercantum dalam kumpulan sumber terbuka untuk Flink sebagai nilai default di Amazon EMR, jadi Anda tidak perlu mengonfigurasinya sendiri.

Kontainer aplikasi Flink membuat dan menulis hingga tiga jenis file log: .out file, .log file, dan .err file. Hanya .err file yang dimampatkan dan dikeluarkan dari sistem fail, sementara .log dan .out file log tetap di sistem file. Untuk memastikan file keluaran ini tetap dapat dikelola dan cluster tetap stabil, Anda dapat mengonfigurasi rotasi log log4j.properties untuk mengatur jumlah file maksimum dan membatasi ukurannya.

Amazon EMR versi 5.30.0 dan yang lebih baru

Dimulai dengan Amazon EMR 5.30.0, Flink menggunakan log4j2 logging framework dengan nama flink-log4j. klasifikasi konfigurasi Contoh konfigurasi konfigurasi Konfigurasi berikut menunjukkan format log4j2.

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

Amazon EMR versi 5.29.0 dan sebelumnya

Dengan Amazon EMR versi 5.29.0 dan sebelumnya, Flink menggunakan kerangka log4j logging. Contoh konfigurasi berikut menunjukkan format 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"
    },
  }
]

Amazon EMR merilis 6.12.0 dan yang lebih tinggi memberikan dukungan runtime Java 11 untuk Flink. Bagian berikut menjelaskan cara mengkonfigurasi cluster untuk menyediakan dukungan runtime Java 11 untuk Flink.

Gunakan langkah-langkah berikut untuk membuat cluster EMR dengan runtime Flink dan Java 11. File konfigurasi tempat Anda menambahkan dukungan runtime Java 11 adalahflink-conf.yaml.

Console
Untuk membuat cluster dengan runtime Flink dan Java 11 di konsol

  1. Masuk ke AWS Management Console, dan buka konsol EMR Amazon di https://console.aws.amazon.com /emr.

  2. Pilih Cluster di bawah EMR EC2 di panel navigasi, lalu Buat cluster.

  3. Pilih Amazon EMR rilis 6.12.0 atau lebih tinggi, dan pilih untuk menginstal aplikasi Flink. Pilih aplikasi lain yang ingin Anda instal di cluster Anda.

  4. Lanjutkan menyiapkan cluster Anda. Di bagian Pengaturan Perangkat Lunak opsional, gunakan opsi Enter konfigurasi default dan masukkan konfigurasi berikut:

    [
    {
      "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"
      }
    }
]

  5. Lanjutkan untuk mengatur dan meluncurkan cluster Anda.

AWS CLI
Untuk membuat cluster dengan runtime Flink dan Java 11 dari CLI

  1. Buat file konfigurasi configurations.json yang mengkonfigurasi Flink untuk menggunakan Java 11. 

    [
    {
      "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"
      }
    }
]

  2. Dari AWS CLI, buat cluster EMR baru dengan Amazon EMR rilis 6.12.0 atau lebih tinggi, dan instal aplikasi Flink, seperti yang ditunjukkan pada contoh berikut:

    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

Gunakan langkah-langkah berikut untuk memperbarui cluster EMR yang sedang berjalan dengan runtime Flink dan Java 11. File konfigurasi tempat Anda menambahkan dukungan runtime Java 11 adalahflink-conf.yaml.

Console
Untuk memperbarui cluster yang sedang berjalan dengan runtime Flink dan Java 11 di konsol

  1. Masuk ke AWS Management Console, dan buka konsol EMR Amazon di https://console.aws.amazon.com /emr.

  2. Pilih Cluster di bawah EMR EC2 di panel navigasi, lalu pilih cluster yang ingin Anda perbarui.

    catatan

    Cluster harus menggunakan Amazon EMR rilis 6.12.0 atau lebih tinggi untuk mendukung Java 11.

  3. Pilih tab Konfigurasi.

  4. Di bagian Konfigurasi grup instans, pilih grup Running instance yang ingin Anda perbarui, lalu pilih Konfigurasi ulang dari menu tindakan daftar.

  5. Konfigurasikan ulang grup instance dengan opsi Edit atribut sebagai berikut. Pilih Tambahkan konfigurasi baru setelah masing-masing.

    Klasifikasi Properti Nilai

    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

  6. Pilih Simpan perubahan untuk menambahkan konfigurasi.

AWS CLI
Untuk memperbarui cluster yang sedang berjalan untuk menggunakan runtime Flink dan Java 11 dari CLI

Gunakan perintah modify-instance-groups untuk menentukan konfigurasi baru untuk grup instans di klaster yang sedang berjalan.

  1. Pertama, buat file konfigurasi configurations.json yang mengkonfigurasi Flink untuk menggunakan Java 11. Dalam contoh berikut, ganti ig-1xxxxxxx9 dengan ID untuk grup instance yang ingin Anda konfigurasi ulang. Simpan file di direktori yang sama tempat Anda akan menjalankan modify-instance-groups perintah.

    [
   {
      "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":[]
         }
      ]
   }
]

  2. Dari AWS CLI, jalankan perintah berikut. Ganti ID untuk grup instance yang ingin Anda konfigurasi ulang:

    aws emr modify-instance-groups --cluster-id j-2AL4XXXXXX5T9 \
--instance-groups file://configurations.json

Untuk menentukan runtime Java untuk cluster yang sedang berjalan, masuk ke node utama dengan SSH seperti yang dijelaskan dalam Connect to the primary node with SSH. Kemudian jalankan perintah berikut:

ps -ef | grep flink

psPerintah dengan -ef opsi mencantumkan semua proses yang berjalan pada sistem. Anda dapat memfilter output itu grep untuk menemukan sebutan stringflink. Tinjau output untuk nilai Java Runtime Environment (JRE),. jre-XX Dalam output berikut, jre-11 menunjukkan bahwa Java 11 diambil saat runtime untuk Flink.

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.

Atau, masuk ke node utama dengan SSH dan mulai sesi Flink YARN dengan perintah. flink-yarn-session -d Output menunjukkan Java Virtual Machine (JVM) untuk Flink, java-11-amazon-corretto dalam contoh berikut:

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