Requisiti per il committer ottimizzato per EMRFS S3

Il seguente esempio di Scala indica a Spark di utilizzare un algoritmo di commit diverso, che impedisce del tutto l'uso del committer ottimizzato per S3. EMRFS Il codice imposta la proprietà partitionOverwriteMode su dynamic per sovrascrivere solo le partizioni su cui si stanno scrivendo i dati. Quindi, le colonne delle partizioni dinamiche vengono specificate da partitionBy e la modalità scrittura è impostata su 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://amzn-s3-demo-bucket1/output")

È necessario configurare tutte e tre le impostazioni per evitare di utilizzare il committer ottimizzato per S3. EMRFS In questo modo, Spark esegue un algoritmo di commit diverso specificato nel protocollo di commit di Spark. Per le versioni di Amazon EMR 5.x precedenti alla 5.30.0 e per le versioni di Amazon EMR 6.x precedenti alla 6.2.0, il protocollo commit utilizza la staging directory di Spark, che è una directory temporanea creata nella posizione di output che inizia con. .spark-staging L'algoritmo rinomina in modo sequenziale le directory delle partizioni e questo può influire negativamente sulle prestazioni. Per ulteriori informazioni sulle EMR versioni di Amazon 5.30.0 e successive e 6.2.0 e successive, consulta. Usa il protocollo di commit ottimizzato per S3 EMRFS

L'algoritmo in Spark 2.4.0 segue questi passaggi:

In questo esempio, il codice Scala viene inserito in due partizioni. Una partizione ha una posizione di partizione personalizzata. L'altra partizione usa il percorso di partizione predefinito. Il committer EMRFS ottimizzato per S3 viene utilizzato solo per scrivere l'output delle attività nella partizione che utilizza la posizione di partizione predefinita.

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)

Il codice Scala crea i seguenti oggetti 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$

Quando si scrive nelle partizioni in posizioni personalizzate, Spark utilizza un algoritmo di commit simile all'esempio precedente, che è descritto di seguito. Come nell'esempio precedente, l'algoritmo produce ridenominazioni sequenziali che possono influire negativamente sulle prestazioni.