View a markdown version of this page

Persyaratan untuk Committer EMRFS S3-optimized - Amazon EMR
Kesempatan ketika komit EMRFS S3-optimized tidak digunakan

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

Persyaratan untuk Committer EMRFS S3-optimized

S3-optimized Committer EMRFS digunakan ketika kondisi berikut terpenuhi:

  • Anda menjalankan tugas Spark yang menggunakan Spark DataFrames,, atau Datasets untuk menulis file ke Amazon S3. Dimulai dengan Amazon EMR 6.4.0, commit ini dapat digunakan untuk semua format umum termasuk parket, ORC, dan format berbasis teks (termasuk CSV dan JSON). Untuk rilis sebelum Amazon EMR 6.4.0, hanya format Parquet yang didukung.

  • Multipart upload diaktifkan di Amazon EMR. Ini adalah opsi default. Untuk informasi selengkapnya, lihat Penguat EMRFS dan ungg S3-optimized ahan multibagian.

  • Dukungan format file bawaan Spark digunakan. Dukungan format file bawaan digunakan dalam keadaan berikut:

    • Untuk tabel metastore Hive, bila spark.sql.hive.convertMetastoreParquet disetel ke true untuk tabel Parquet, atau spark.sql.hive.convertMetastoreOrc disetel ke true untuk tabel Orc dengan Amazon EMR 6.4.0 atau lebih tinggi. Ini adalah pengaturan default.

    • Ketika pekerjaan menulis ke sumber data format file atau tabel—misalnya, tabel target dibuat dengan klausa. USING parquet

    • Ketika pekerjaan menulis ke tabel Parket non-dipartisi Hive metastore. built-in dukungan Parket Spark tidak mendukung tabel Hive dipartisi, yang merupakan keterbatasan diketahui. Untuk informasi lebih lanjut, lihat Kon versi tabel Hive metastore Parquet di Apache Spark, DataFrames dan Panduan Kumpulan Data.

  • Spark pekerjaan operasi yang menulis ke lokasi partisi default — misalnya, ${table_location}/k1=v1/k2=v2/—gunakan committer. Committer tidak digunakan jika operasi pekerjaan menulis ke lokasi partisi kustom—misalnya, jika lokasi partisi kustom disetel menggunakan ALTER TABLE SQL perintah.

  • Nilai berikut untuk Spark mesti digunakan:

    • Parameter spark.sql.parquet.fs.optimized.committer.optimization-enabled properti harus diatur ke true. Ini adalah pengaturan default dengan Amazon EMR 5.20.0 dan kemudian. Dengan Amazon EMR 5.19.0, nilai default adalah false. Untuk informasi tentang mengonfigurasi retensi, lihat Aktifkan komit EMRFS untuk Amazon S3-optimized EMR 5.19.0.

    • Jika menulis ke tabel metastore Hive yang tidak dipartisi, hanya format file Parquet dan Orc yang didukung. spark.sql.hive.convertMetastoreParquetharus disetel ke true jika menulis ke tabel metastore Parquet Hive yang tidak dipartisi. spark.sql.hive.convertMetastoreOrcharus disetel ke true jika menulis ke tabel metastore Orc Hive yang tidak dipartisi. Ini adalah pengaturan default.

    • spark.sql.parquet.output.committer.class harus diatur ke com.amazon.emr.committer.EmrOptimizedSparkSqlParquetOutputCommitter. Ini adalah pengaturan default.

    • spark.sql.sources.commitProtocolClassharus diatur ke org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol atauorg.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol. org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocoladalah pengaturan default untuk Amazon EMR 5.x seri versi 5.30.0 dan lebih tinggi, dan untuk Amazon EMR 6.x seri versi 6.2.0 dan lebih tinggi. org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocoladalah pengaturan default untuk versi Amazon EMR sebelumnya.

    • Jika pekerjaan Spark menimpa dataset Parket dipartisi dengan kolom partisi dinamis, maka partitionOverwriteMode tulis opsi dan spark.sql.sources.partitionOverwriteMode harus diatur ke static. Ini adalah pengaturan default.

      catatan

      Parameter partitionOverwriteMode menulis opsi diperkenalkan di Spark 2.4.0. Untuk Spark versi 2.3.2, disertakan dengan rilis Amazon EMR 5.19.0, mengatur spark.sql.sources.partitionOverwriteMode properti.

Kesempatan ketika komit EMRFS S3-optimized tidak digunakan

Umumnya, commit EMRFS S3-optimized tidak digunakan dalam situasi berikut.

Situasi Mengapa Committer tidak digunakan
Saat Anda menulis ke HDFS Committer hanya mendukung penulisan ke Amazon S3 menggunakan EMRFS.
Saat Anda menggunakan sistem file S3A Committer hanya mendukung EMRFS.
Saat Anda menggunakan MapReduce atau API RDD Spark Committer hanya mendukung penggunaan SparkSQL, DataFrame, atau Dataset API.

Contoh Scala berikut menunjukkan beberapa situasi tambahan yang mencegah commit EMRFS S3-optimized digunakan secara keseluruhan (contoh pertama) dan sebagian (contoh kedua).

contoh- Mode penggantian partisi dinamis

Contoh Scala berikut menginstruksikan Spark untuk menggunakan algoritma komit yang berbeda, yang mencegah penggunaan commit EMRFS sama sekali S3-optimized . Kode menetapkan partitionOverwriteMode properti dynamic untuk mengganti hanya partisi yang Anda tulis data. Kemudian, kolom partisi dinamis ditentukan olehpartitionBy, dan mode tulis diatur keoverwrite. 

val dataset = spark.range(0, 10)
  .withColumn("dt", expr("date_sub(current_date(), id)"))

dataset.write.mode("overwrite")
  .option("partitionOverwriteMode", "dynamic")
  .partitionBy("dt")
  .parquet("s3://amzn-s3-demo-bucket1/output")

Anda harus mengkonfigurasi ketiga pengaturan untuk menghindari penggunaan komitter EMRFS S3-optimized . Ketika Anda melakukannya, Spark mengeksekusi algoritma komit berbeda yang ditentukan dalam protokol komit Spark. Untuk rilis Amazon EMR 5.x lebih awal dari 5.30.0 dan untuk rilis Amazon EMR 6.x lebih awal dari 6.2.0, protokol commit menggunakan direktori pementasan Spark, yang merupakan direktori sementara yang dibuat di bawah lokasi keluaran yang dimulai dengan. .spark-staging Algoritma secara berurutan mengganti nama direktori partisi, yang dapat berdampak negatif pada kinerja. Untuk informasi selengkapnya tentang rilis Amazon EMR 5.30.0 dan yang lebih baru dan 6.2.0 dan yang lebih baru, lihat. Gunakan protokol komit EMRFS S3-optimized

Algoritma di Spark 2.4.0 mengikuti langkah-langkah berikut:

  1. Upaya tugas menulis output mereka ke partisi direktori di bawah direktori pementasan Spark — misalnya, ${outputLocation}/spark-staging-${jobID}/k1=v1/k2=v2/.

  2. Untuk setiap partisi yang ditulis, percobaan tugas terus melacak path partisi relatif—misalnya, k1=v1/k2=v2.

  3. Ketika suatu tugas berhasil diselesaikan, tugas terkait menyediakan semua jalur partisi secara relatif yang dilacaknya kepada driver.

  4. Setelah semua tugas selesai, pekerjaan commit fase mengumpulkan semua direktori partisi yang berhasil tugas mencoba menulis di bawah direktori pementasan Spark ini. Spark berurutan mengganti nama masing-masing direktori ini ke lokasi output akhir menggunakan pohon direktori mengubah nama operasi.

  5. Direktori pementasan dihapus sebelum pekerjaan komit fase selesai.

contoh— Lokasi partisi khusus

Dalam contoh ini, kode Scala menyisipkan dalam dua partisi. Satu partisi memiliki lokasi partisi kustom. Partisi lain menggunakan lokasi partisi default. S3-optimized Committer EMRFS hanya digunakan untuk menulis output tugas ke partisi yang menggunakan lokasi partisi default.


val table = "dataset"
val location = "s3://bucket/table"
                            
spark.sql(s"""
  CREATE TABLE $table (id bigint, dt date) 
  USING PARQUET PARTITIONED BY (dt) 
  LOCATION '$location'
""")
                            
// Add a partition using a custom location
val customPartitionLocation = "s3://bucket/custom"
spark.sql(s"""
  ALTER TABLE $table ADD PARTITION (dt='2019-01-28') 
  LOCATION '$customPartitionLocation'
""")
                            
// Add another partition using default location
spark.sql(s"ALTER TABLE $table ADD PARTITION (dt='2019-01-29')")
                            
def asDate(text: String) = lit(text).cast("date")
                            
spark.range(0, 10)
  .withColumn("dt",
    when($"id" > 4, asDate("2019-01-28")).otherwise(asDate("2019-01-29")))
  .write.insertInto(table)

Kode Scala menciptakan objek Amazon S3 berikut:

custom/part-00001-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet
custom_$folder$
table/_SUCCESS
table/dt=2019-01-29/part-00000-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet
table/dt=2019-01-29_$folder$
table_$folder$

Ketika menulis ke partisi di lokasi kustom, Spark menggunakan algoritma komit mirip dengan contoh sebelumnya, yang diuraikan di bawah ini. Seperti contoh sebelumnya, algoritma menghasilkan penggantian nama berurutan, yang dapat berdampak negatif pada kinerja.

  1. Ketika menulis output ke partisi di lokasi kustom, tugas menulis ke file di bawah direktori pementasan Spark ini, yang dibuat di bawah lokasi output akhir. Nama file termasuk UUID acak untuk melindungi terhadap tabrakan file. Upaya tugas melacak setiap file bersama dengan path output akhir yang diinginkan.

  2. Ketika tugas selesai berhasil, menyediakan driver dengan file dan akhir yang diinginkan output jalan mereka.

  3. Setelah semua tugas selesai, pekerjaan commit fase berurutan mengganti nama semua file yang ditulis untuk partisi di lokasi kustom ke jalur output akhir mereka.

  4. Direktori pementasan dihapus sebelum pekerjaan komit fase selesai.