EMRFS S3-optimized遞交者的需求

下列 Scala 範例會指示 Spark 使用不同的遞交演算法，以防止完全使用 EMRFS S3-optimized遞交者。程式碼將 partitionOverwriteMode 屬性設為 dynamic，僅覆寫您要寫入資料的分割區。然後，由 partitionBy 指定動態分割區資料欄，且寫入模式被設為 overwrite。

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://EXAMPLE-DOC-BUCKET/output")

您必須設定所有三個設定，以避免使用 EMRFS S3-optimized遞交者。當您這樣做時，Spark 會執行在 Spark 遞交通訊協定中指定的其他遞交演算法。對於 5.30.0 之前的 Amazon EMR 5.x 版本，以及 6.2.0 EMR 之前的 Amazon 6.x 版本，遞交通訊協定會使用 Spark 的暫存目錄，這是在以 開頭的輸出位置下建立的暫存目錄.spark-staging。該演算法會按順序重新命名分割區目錄，這可能對效能產生負面影響。如需 Amazon 5.30.0 及更新EMR版本以及 6.2.0 及更新版本的詳細資訊，請參閱 使用 EMRFS S3-optimized遞交通訊協定。

Spark 2.4.0 中的演算法遵循以下步驟：

在此範例中，該 Scala 程式碼將插入至兩個分割區。其中一個分割區有自訂的分割區位置。另一個分割區使用預設分割區位置。EMRFS S3-optimized遞交器僅用於將任務輸出寫入使用預設分割區位置的分割區。

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)

Scala 程式碼會建立以下 Amazon S3 物件：

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$

在寫入至自訂位置的分割區時，Spark 會使用一個和上一個範例相似的遞交演算法，如下所述。如之前的範例所示，該演算法會導致順序重新命名，這可能會對效能產生負面影響。