我正在嘗試使用 s3 的不同 Spark 輸出提交器設定，并想嘗試使用魔術提交器。到目前為止，我還沒有設法讓我的作業使用魔法提交器，而且它們似乎總是依賴于檔案輸出提交器。
我正在運行的 Spark 作業是一個簡單的 PySpark 測驗作業，它運行一個簡單的查詢、重新磁區資料并將 parquet 輸出到 s3：
df = spark.sql("select * from some_table where some_condition")
df.write \
.partitionBy("some_column") \
.parquet("s3://some-bucket/some-folder", mode="overwrite")
相關的 spark 設定是（取自 Spark UI，作業的環境選項卡）：
spark.hadoop.mapreduce.outputcommitter.factory.scheme.s3a org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory
spark.hadoop.fs.s3a.committer.magic.enabled true
spark.hadoop.fs.s3a.committer.name magic
spark.hadoop.fs.s3a.committer.staging.tmp.path tmp/staging
spark.hadoop.fs.s3a.committer.staging.unique-filenames true
spark.sql.parquet.output.committer.class org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter
spark.sql.sources.commitProtocolClass org.apache.spark.internal.io.cloud.PathOutputCommitProtocol
mapreduce.output.fileoutputformat.compress false
mapreduce.output.fileoutputformat.compress.codec org.apache.hadoop.io.compress.DefaultCodec
mapreduce.output.fileoutputformat.compress.type RECORD
mapreduce.outputcommitter.factory.scheme.s3a org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory
mapreduce.fileoutputcommitter.algorithm.version 1
mapreduce.fileoutputcommitter.task.cleanup.enabled false
mapreduce.outputcommitter.factory.scheme.s3a org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory
Hadoop 屬性：
fs.s3a.committer.magic.enabled true
fs.s3a.committer.name magic
（讓我知道是否有任何其他設定相關）
我基于對使用檔案提交者而不是魔術提交者的觀察基于以下幾點：
- spark作業生成的不同日志行似乎表明正在使用的檔案輸出提交者：
"class":"org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter","file_line":"FileOutputCommitter.java:601","func":"commitTask","message":"Saved output of task 'attempt_2021...' to s3://some-bucket/some-folder/_temporary/0/
task_2021..."
"class":"org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat","file_line":"ParquetFileFormat.scala:54","message":"U
sing user defined output committer for Parquet: org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter"
"class":"org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter","file_line":"FileOutputCommitter.java:141","func":"<init>","message":"File Outpu
t Committer Algorithm version is 1"
"class":"org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter","file_line":"FileOutputCommitter.java:156","func":"<init>","message":"FileOutput
Committer skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false"
- 將檔案提交者的演算法設定為無效數字時，如下所示：
spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version -7
從檔案提交者的建構式中引發例外，表示該值無效 - 暗示檔案提交者已初始化而不是魔術提交者。
我沒有看到任何指示使用魔術提交者的日志，也沒有看到任何初始化提交者失敗的日志，這可以解釋回退到檔案提交者。
Spark 版本是 3.1.2 使用這個 spark-hadoop-cloud JAR。讓我知道是否有任何其他正式發布的 JAR 我可以嘗試，或者是否有任何其他可能相關的日志指示。
有什么想法嗎？
====== 編輯：
下面是我在將檔案提交者演算法設定為無效值時看到的堆疊跟蹤。似乎呼叫org.apache.spark.internal.io.cloud.PathOutputCommitProtocol.setupCommitter最終呼叫org.apache.hadoop.mapreduce.lib.output.FileOutputCommitterFactory.createOutputCommitter它又初始化了不正確的型別org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter而不是配置的型別org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter
Py4JJavaError: An error occurred while calling o259.parquet.
: java.io.IOException: Only 1 or 2 algorithm version is supported
at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.<init>(FileOutputCommitter.java:143)
at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.<init>(FileOutputCommitter.java:117)
at org.apache.hadoop.mapreduce.lib.output.PathOutputCommitterFactory.createFileOutputCommitter(PathOutputCommitterFactory.java:134)
at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitterFactory.createOutputCommitter(FileOutputCommitterFactory.java:35)
at org.apache.hadoop.mapreduce.lib.output.PathOutputCommitterFactory.createCommitter(PathOutputCommitterFactory.java:201)
at org.apache.spark.internal.io.cloud.PathOutputCommitProtocol.setupCommitter(PathOutputCommitProtocol.scala:88)
at org.apache.spark.internal.io.cloud.PathOutputCommitProtocol.setupCommitter(PathOutputCommitProtocol.scala:49)
at org.apache.spark.internal.io.HadoopMapReduceCommitProtocol.setupJob(HadoopMapReduceCommitProtocol.scala:177)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:173)
at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala:188)
at org.apache.spark.sql.execution.command.DataWritingCommandExec.sideEffectResult$lzycompute(commands.scala:108)
at org.apache.spark.sql.execution.command.DataWritingCommandExec.sideEffectResult(commands.scala:106)
at org.apache.spark.sql.execution.command.DataWritingCommandExec.doExecute(commands.scala:131)
at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180)
at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176)
at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:132)
at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:131)
at org.apache.spark.sql.DataFrameWriter.$anonfun$runCommand$1(DataFrameWriter.scala:989)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:103)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:163)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:90)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:989)
at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:438)
at org.apache.spark.sql.DataFrameWriter.saveInternal(DataFrameWriter.scala:415)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:293)
at org.apache.spark.sql.DataFrameWriter.parquet(DataFrameWriter.scala:874)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:748)
uj5u.com熱心網友回復：
謎團已解決 - 無法初始化魔術提交者是由于提交者工廠方案設定與實際目標 URL 的方案不匹配。考慮一下：
提交者工廠配置是使用密鑰設定的：spark.hadoop.mapreduce.outputcommitter.factory.scheme.s3a- 意味著該設定是為 s3a 協議 URL 進行的。
雖然發送到 write 方法的 URL 是：s3://some-bucket/some-folder- 使用 s3 協議而不是 s3a。
在PathOutputCommitterFactory與模式的配置關鍵的Hadoop類搜索mapreduce.outputcommitter.factory.scheme.%s能夠識別用于給定輸出URL哪個工廠。如果配置鍵中設定的模式（在這種情況下s3a）與目標 URL 中的模式（在這種情況下）不匹配s3- 提交者工廠設定將無法識別，工廠型別將回退FileOutputCommitter。
解決方案 - 確保outputcommitter.factory.scheme.<protocol>設定與目標 URL 中的協議相匹配。我已經成功地在 URL 和配置鍵中使用s3和s3a進行了測驗。
uj5u.com熱心網友回復：
這聽起來像是一個系結問題，但我無法立即看到它在哪里。一目了然，您擁有所有正確的設定。
檢查是否正在使用 S3 委員會的最簡單方法是查看 _SUCCESS 檔案。如果它是一段 JSON，那么就使用了一個新的提交者……里面的文本會告訴你更多關于提交者的資訊。
一個 0 位元組的檔案意味著仍然使用經典的檔案輸出提交者
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/401930.html