对 EMRFS S3 优化的提交者的要求

以下 Scala 示例指示 Spark 使用不同的提交算法，这样可以完全阻止使用 EMRFS S3 优化的提交器。该代码将 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 优化的提交器。当您执行此操作时，Spark 会执行在 Spark 的提交协议中指定的不同提交算法。对于 EMR 5.30.0 之前的 Amazon 5.x 版本和 EMR 6.2.0 之前的 Amazon 6.x 版本，提交协议使用 Spark 的暂存目录，该目录是在以开头的输出位置下创建的临时目录。.spark-staging该算法按顺序对分区目录进行重命名，这可能会对性能产生负面影响。有关 Amazon 5.30.0 及更高EMR版本以及 6.2.0 及更高版本的更多信息，请参阅。使用 EMRFS S3 优化的提交协议

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

在此示例中，Scala 代码插入到两个分区中。一个分区具有自定义分区位置。另一个分区使用默认分区位置。EMRFSS3 优化的提交器仅用于将任务输出写入使用默认分区位置的分区。

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 会使用类似于上一个示例的提交算法，如下所述。与前面的示例一样，该算法会导致顺序重命名，这可能会影响性能。