You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Sayed Mohammad Hossein Torabi (Jira)" <ji...@apache.org> on 2023/11/05 23:28:00 UTC
[jira] [Updated] (SPARK-45800) FileOutputCommiter race condition
[ https://issues.apache.org/jira/browse/SPARK-45800?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Sayed Mohammad Hossein Torabi updated SPARK-45800:
--------------------------------------------------
Attachment: main.py
> FileOutputCommiter race condition
> ---------------------------------
>
> Key: SPARK-45800
> URL: https://issues.apache.org/jira/browse/SPARK-45800
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 3.0.3
> Reporter: Sayed Mohammad Hossein Torabi
> Priority: Major
> Attachments: main.py
>
>
> Race condition happens when multiple spark jobs write to a table simultaneously on the *FileOutputCommiter* side
> Here is an example:
> {code:sql}
> import tempfile
> from concurrent.futures import ThreadPoolExecutor
> from pyspark.sql import SparkSession
> from pyspark.sql.functions import lit, rand
> from pyspark.sql.types import IntegerType
> TABLE_NAME = "test_table"
> table_directory = tempfile.TemporaryDirectory()
> spark = (
> SparkSession.builder.master("local[*]")
> .config("spark.sql.shuffle.partition", 1)
> .config("spark.driver.host", "localhost")
> .config("spark.sql.session.timeZone", "Europe/Amsterdam")
> .config("spark.sql.source.partitionOverwriteMode", "dynamic")
> .enableHiveSupport()
> .getOrCreate()
> )
> spark.sql(
> f"""
> CREATE EXTERNAL TABLE IF NOT EXISTS {TABLE_NAME} (
> value INT
> )
> PARTITIONED BY (
> date STRING
> )
> STORED AS PARQUET
> LOCATION '{table_directory.name}'
> TBLPROPERTIES (
> "parquet.compress"="SNAPPY"
> )
> """
> )
> with ThreadPoolExecutor(max_workers=3) as executor:
> executor.map(
> lambda date: (
> spark.range(1e6)
> .withColumn("value", (rand(seed=42) * 100).cast(IntegerType()))
> .withColumn("date", lit(date))
> .select("value", "date")
> .write.insertInto(TABLE_NAME, True)
> ),
> [
> "1970-01-01",
> "1970-01-02",
> "1970-01-03",
> ],
> )
> spark.sql(f"DROP TABLE {TABLE_NAME}")
> table_directory.cleanup()
> spark.stop(){code}
> Log:
> {code:java}
> 3/11/06 00:12:44 WARN HiveConf: HiveConf of name hive.stats.jdbc.timeout does not exist
> 23/11/06 00:12:44 WARN HiveConf: HiveConf of name hive.stats.retries.wait does not exist
> 23/11/06 00:12:45 WARN ObjectStore: Version information not found in metastore. hive.metastore.schema.verification is not enabled so recording the schema version 2.3.0
> 23/11/06 00:12:45 WARN ObjectStore: setMetaStoreSchemaVersion called but recording version is disabled: version = 2.3.0, comment = Set by MetaStore blcksrx@127.0.0.1
> 23/11/06 00:12:45 WARN SessionState: METASTORE_FILTER_HOOK will be ignored, since hive.security.authorization.manager is set to instance of HiveAuthorizerFactory.
> 23/11/06 00:12:45 WARN HiveConf: HiveConf of name hive.internal.ss.authz.settings.applied.marker does not exist
> 23/11/06 00:12:45 WARN HiveConf: HiveConf of name hive.stats.jdbc.timeout does not exist
> 23/11/06 00:12:45 WARN HiveConf: HiveConf of name hive.stats.retries.wait does not exist
> 23/11/06 00:12:45 WARN HiveConf: HiveConf of name hive.internal.ss.authz.settings.applied.marker does not exist
> 23/11/06 00:12:45 WARN HiveConf: HiveConf of name hive.stats.jdbc.timeout does not exist
> 23/11/06 00:12:45 WARN HiveConf: HiveConf of name hive.stats.retries.wait does not exist
> 23/11/06 00:12:45 WARN HiveConf: HiveConf of name hive.internal.ss.authz.settings.applied.marker does not exist
> 23/11/06 00:12:45 WARN HiveConf: HiveConf of name hive.stats.jdbc.timeout does not exist
> 23/11/06 00:12:45 WARN HiveConf: HiveConf of name hive.stats.retries.wait does not exist
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 96.54% for 7 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 84.47% for 8 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 75.08% for 9 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 67.58% for 10 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 75.08% for 9 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 84.47% for 8 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 96.54% for 7 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 96.54% for 7 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 84.47% for 8 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 75.08% for 9 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 67.58% for 10 writers
> 23/11/06 00:12:47 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 75.08% for 9 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 84.47% for 8 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 96.54% for 7 writers
> 23/11/06 00:12:47 ERROR FileFormatWriter: Aborting job 5520a0d3-cc99-4caa-9b67-4f707373e132.
> java.io.FileNotFoundException: File file:/var/folders/vc/dk5pd3b95g73wx_v59q46z400000gn/T/tmpkvty8p4b/_temporary/0 does not exist
> at org.apache.hadoop.fs.RawLocalFileSystem.listStatus(RawLocalFileSystem.java:489)
> at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1868)
> at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1910)
> at org.apache.hadoop.fs.ChecksumFileSystem.listStatus(ChecksumFileSystem.java:678)
> at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1868)
> at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1910)
> at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.getAllCommittedTaskPaths(FileOutputCommitter.java:332)
> at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.commitJobInternal(FileOutputCommitter.java:402)
> at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.commitJob(FileOutputCommitter.java:375)
> at org.apache.parquet.hadoop.ParquetOutputCommitter.commitJob(ParquetOutputCommitter.java:48)
> at org.apache.spark.internal.io.HadoopMapReduceCommitProtocol.commitJob(HadoopMapReduceCommitProtocol.scala:167)
> at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:220)
> at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala:178)
> 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:127)
> at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:126)
> at org.apache.spark.sql.DataFrameWriter.$anonfun$runCommand$1(DataFrameWriter.scala:962)
> at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100)
> at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160)
> at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87)
> at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:767)
> at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
> at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:962)
> at org.apache.spark.sql.DataFrameWriter.insertInto(DataFrameWriter.scala:518)
> at org.apache.spark.sql.DataFrameWriter.insertInto(DataFrameWriter.scala:473)
> 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:750)
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 96.54% for 7 writers
> 23/11/06 00:12:47 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 96.54% for 7 writers
> 23/11/06 00:12:47 WARN log: Updating partition stats fast for: test_table
> 23/11/06 00:12:48 WARN log: Updated size to 885865
> 23/11/06 00:12:48 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 96.54% for 7 writers
> 23/11/06 00:12:48 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 84.47% for 8 writers
> 23/11/06 00:12:48 WARN MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
> Scaling row group sizes to 96.54% for 7 writers
> 23/11/06 00:12:48 ERROR FileFormatWriter: Aborting job 191bc7bd-7729-4983-9f19-6d2d5e1f534b.
> java.io.FileNotFoundException: File file:/var/folders/vc/dk5pd3b95g73wx_v59q46z400000gn/T/tmpkvty8p4b/_temporary/0 does not exist
> at org.apache.hadoop.fs.RawLocalFileSystem.listStatus(RawLocalFileSystem.java:489)
> at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1868)
> at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1910)
> at org.apache.hadoop.fs.ChecksumFileSystem.listStatus(ChecksumFileSystem.java:678)
> at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1868)
> at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1910)
> at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.getAllCommittedTaskPaths(FileOutputCommitter.java:332)
> at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.commitJobInternal(FileOutputCommitter.java:402)
> at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.commitJob(FileOutputCommitter.java:375)
> at org.apache.parquet.hadoop.ParquetOutputCommitter.commitJob(ParquetOutputCommitter.java:48)
> at org.apache.spark.internal.io.HadoopMapReduceCommitProtocol.commitJob(HadoopMapReduceCommitProtocol.scala:167)
> at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:220)
> at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala:178)
> 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:127)
> at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:126)
> at org.apache.spark.sql.DataFrameWriter.$anonfun$runCommand$1(DataFrameWriter.scala:962)
> at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100)
> at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160)
> at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87)
> at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:767)
> at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
> at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:962)
> at org.apache.spark.sql.DataFrameWriter.insertInto(DataFrameWriter.scala:518)
> at org.apache.spark.sql.DataFrameWriter.insertInto(DataFrameWriter.scala:473)
> 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:750)
> {code}
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org