[ https://issues.apache.org/jira/browse/SPARK-46990?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17828090#comment-17828090 ]
Pavlo Pohrrebnyi commented on SPARK-46990: ------------------------------------------ We are experiencing the same with Spark 3.5. That is likely caused by [SPARK-46633|https://issues.apache.org/jira/browse/SPARK-46633]. Here is the change [PR-[44635|https://github.com/apache/spark/pull/44635/files#diff-c139f61eabcfcb9725c8caeb747becae061a2ea44f774b12c9cce5aeac102880]|[https://github.com/apache/spark/pull/44635/files#diff-c139f61eabcfcb9725c8caeb747becae061a2ea44f774b12c9cce5aeac102880]][.|https://github.com/apache/spark/pull/44635/files#diff-c139f61eabcfcb9725c8caeb747becae061a2ea44f774b12c9cce5aeac102880] The job hangs once tries to read avro files with no records. It loops forever here: {code:java} org.apache.spark.sql.avro.AvroUtils$RowReader.hasNextRow(AvroUtils.scala:265) org.apache.spark.sql.avro.AvroUtils$RowReader.hasNextRow$(AvroUtils.scala:263) org.apache.spark.sql.avro.AvroFileFormat$$anon$1.hasNextRow(AvroFileFormat.scala:186) org.apache.spark.sql.avro.AvroFileFormat$$anon$1.hasNext(AvroFileFormat.scala:201) scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1$$anon$2.getNext(FileScanRDD.scala:604) org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73) org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:798) org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.$anonfun$hasNext$1(FileScanRDD.scala:506) org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1$$Lambda$1269/1728924771.apply$mcZ$sp(Unknown Source) {code} Here is the sample to reproduce the issue: [^second=02.avro] > Regression: Unable to load empty avro files emitted by event-hubs > ----------------------------------------------------------------- > > Key: SPARK-46990 > URL: https://issues.apache.org/jira/browse/SPARK-46990 > Project: Spark > Issue Type: Bug > Components: PySpark > Affects Versions: 3.5.0 > Environment: Databricks 14.0 - 14.3 (spark 3.5.0) > Reporter: Kamil Kandzia > Priority: Major > Attachments: second=02.avro > > > In azure, I use databricks and event-hubs. Up until spark version 3.4.1 (in > databricks as 13.3 LTS) empty avro files emitted by event-hubs can be read. > Since version 3.5.0, it is impossible to load these files (even if I have > multiple avro files to load and one of them is empty, it can't perform an > operation like count or save). I tested this on databricks versions 14.0, > 14.1, 14.2, 14.3 and it doesn't work properly in any of them. > I use the following code: > > {code:java} > df = spark.read.format("avro") \ > .load('abfss://<container>@<storage>.dfs.core.windows.net/<evh-namespace>/<evh>/0/2024/02/05/22/46/10.avro') > > df.count() <- in this operation the spark hangs{code} > I am sending a fragment of logs from databricks and query plan: > {code:java} > 24/02/06 10:03:10 INFO ProgressReporter$: Added result fetcher for > 2734305632140666820_7640723027790427455_4f56f528d4a44796a98821713778d5f9 > 24/02/06 10:03:11 INFO InMemoryFileIndex: Start listing leaf files and > directories. Size of Paths: 1; threshold: 32 > 24/02/06 10:03:11 INFO InMemoryFileIndex: Start listing leaf files and > directories. Size of Paths: 0; threshold: 32 > 24/02/06 10:03:11 INFO InMemoryFileIndex: It took 9 ms to list leaf files for > 1 paths. > 24/02/06 10:03:11 INFO ProgressReporter$: Removed result fetcher for > 2734305632140666820_7640723027790427455_4f56f528d4a44796a98821713778d5f9 > 24/02/06 10:03:12 INFO ProgressReporter$: Added result fetcher for > 2734305632140666820_6526693737104909881_a07acddb350f44a284cac52db0b2fb21 > 24/02/06 10:03:12 INFO ClusterLoadMonitor: Added query with execution ID:38. > Current active queries:1 > 24/02/06 10:03:12 INFO FileSourceStrategy: Pushed Filters: > 24/02/06 10:03:12 INFO FileSourceStrategy: Post-Scan Filters: > 24/02/06 10:03:12 INFO CodeGenerator: Code generated in 10.636308 ms > 24/02/06 10:03:12 INFO MemoryStore: Block broadcast_34 stored as values in > memory (estimated size 409.5 KiB, free 3.3 GiB) > 24/02/06 10:03:12 INFO MemoryStore: Block broadcast_34_piece0 stored as bytes > in memory (estimated size 14.5 KiB, free 3.3 GiB) > 24/02/06 10:03:12 INFO BlockManagerInfo: Added broadcast_34_piece0 in memory > on <IP_ADDRESS_2>:43781 (size: 14.5 KiB, free: 3.3 GiB) > 24/02/06 10:03:12 INFO SparkContext: Created broadcast 34 from > $anonfun$withThreadLocalCaptured$5 at LexicalThreadLocal.scala:63 > 24/02/06 10:03:12 INFO FileSourceScanExec: Planning scan with bin packing, > max split size: 4194304 bytes, max partition size: 4194304, open cost is > considered as scanning 4194304 bytes. > 24/02/06 10:03:12 INFO DAGScheduler: Registering RDD 104 > ($anonfun$withThreadLocalCaptured$5 at LexicalThreadLocal.scala:63) as input > to shuffle 11 > 24/02/06 10:03:12 INFO DAGScheduler: Got map stage job 22 > ($anonfun$withThreadLocalCaptured$5 at LexicalThreadLocal.scala:63) with 1 > output partitions > 24/02/06 10:03:12 INFO DAGScheduler: Final stage: ShuffleMapStage 31 > ($anonfun$withThreadLocalCaptured$5 at LexicalThreadLocal.scala:63) > 24/02/06 10:03:12 INFO DAGScheduler: Parents of final stage: List() > 24/02/06 10:03:12 INFO DAGScheduler: Missing parents: List() > 24/02/06 10:03:12 INFO DAGScheduler: Submitting ShuffleMapStage 31 > (MapPartitionsRDD[104] at $anonfun$withThreadLocalCaptured$5 at > LexicalThreadLocal.scala:63), which has no missing parents > 24/02/06 10:03:12 INFO DAGScheduler: Submitting 1 missing tasks from > ShuffleMapStage 31 (MapPartitionsRDD[104] at > $anonfun$withThreadLocalCaptured$5 at LexicalThreadLocal.scala:63) (first 15 > tasks are for partitions Vector(0)) > 24/02/06 10:03:12 INFO TaskSchedulerImpl: Adding task set 31.0 with 1 tasks > resource profile 0 > 24/02/06 10:03:12 INFO TaskSetManager: TaskSet 31.0 using PreferredLocationsV1 > 24/02/06 10:03:12 WARN FairSchedulableBuilder: A job was submitted with > scheduler pool 2734305632140666820, which has not been configured. This can > happen when the file that pools are read from isn't set, or when that file > doesn't contain 2734305632140666820. Created 2734305632140666820 with default > configuration (schedulingMode: FIFO, minShare: 0, weight: 1) > 24/02/06 10:03:12 INFO FairSchedulableBuilder: Added task set TaskSet_31.0 > tasks to pool 2734305632140666820 > 24/02/06 10:03:12 INFO TaskSetManager: Starting task 0.0 in stage 31.0 (TID > 449) (<IP_ADDRESS>, executor 3, partition 0, PROCESS_LOCAL, > 24/02/06 10:03:12 INFO MemoryStore: Block broadcast_35 stored as values in > memory (estimated size 137.2 KiB, free 3.3 GiB) > 24/02/06 10:03:12 INFO MemoryStore: Block broadcast_35_piece0 stored as bytes > in memory (estimated size 41.3 KiB, free 3.3 GiB) > 24/02/06 10:03:12 INFO BlockManagerInfo: Added broadcast_35_piece0 in memory > on <IP_ADDRESS_2>:43781 (size: 41.3 KiB, free: 3.3 GiB) > 24/02/06 10:03:12 INFO SparkContext: Created broadcast 35 from broadcast at > TaskSetManager.scala:723 > 24/02/06 10:03:12 INFO BlockManagerInfo: Added broadcast_35_piece0 in memory > on <IP_ADDRESS>:40825 (size: 41.3 KiB, free: 3.6 GiB) > 24/02/06 10:03:13 INFO BlockManagerInfo: Added broadcast_4_piece0 in memory > on <IP_ADDRESS>:40825 (size: 17.6 KiB, free: 3.6 GiB) > 24/02/06 10:03:14 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 0.0, New Ema: 1.0 > 24/02/06 10:03:15 INFO BlockManagerInfo: Added broadcast_34_piece0 in memory > on <IP_ADDRESS>:40825 (size: 14.5 KiB, free: 3.6 GiB) > 24/02/06 10:03:17 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:20 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:23 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:26 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:29 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:32 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:35 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:38 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:41 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:44 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:47 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:50 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:53 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:56 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:03:58 INFO DataSourceFactory$: DataSource Jdbc URL: > jdbc:mariadb://<DELETED FOR JIRA PURPOSES> > 24/02/06 10:03:58 INFO HikariDataSource: metastore-monitor - Starting... > 24/02/06 10:03:58 INFO HikariDataSource: metastore-monitor - Start completed. > 24/02/06 10:03:58 INFO HikariDataSource: metastore-monitor - Shutdown > initiated... > 24/02/06 10:03:58 INFO HikariDataSource: metastore-monitor - Shutdown > completed. > 24/02/06 10:03:58 INFO MetastoreMonitor: Metastore healthcheck successful > (connection duration = 302 milliseconds) > 24/02/06 10:03:59 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:02 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:05 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:08 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:11 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:13 INFO HiveMetaStore: 1: get_database: default > 24/02/06 10:04:13 INFO audit: ugi=root ip=unknown-ip-addr > cmd=get_database: default > 24/02/06 10:04:13 INFO DriverCorral: DBFS health check ok > 24/02/06 10:04:13 INFO DriverCorral: Metastore health check ok > 24/02/06 10:04:14 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:17 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:20 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:23 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:26 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:29 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:32 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:35 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:38 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:41 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:44 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:47 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:50 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:53 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:56 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:04:59 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:02 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:05 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:08 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:11 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:14 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:17 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:20 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:23 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:26 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:29 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:32 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:35 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:38 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > 24/02/06 10:05:41 INFO ClusterLoadAvgHelper: Current cluster load: 1, Old > Ema: 1.0, New Ema: 1.0 > == Parsed Logical Plan == > Relation > [SequenceNumber#451L,Offset#452,EnqueuedTimeUtc#453,SystemProperties#454,Properties#455,Body#456] > avro== Analyzed Logical Plan == > SequenceNumber: bigint, Offset: string, EnqueuedTimeUtc: string, > SystemProperties: > map<string,struct<member0:bigint,member1:double,member2:string,member3:binary>>, > Properties: > map<string,struct<member0:bigint,member1:double,member2:string,member3:binary>>, > Body: binary > Relation > [SequenceNumber#451L,Offset#452,EnqueuedTimeUtc#453,SystemProperties#454,Properties#455,Body#456] > avro== Optimized Logical Plan == > Relation > [SequenceNumber#451L,Offset#452,EnqueuedTimeUtc#453,SystemProperties#454,Properties#455,Body#456] > avro== Physical Plan == > FileScan avro > [SequenceNumber#451L,Offset#452,EnqueuedTimeUtc#453,SystemProperties#454,Properties#455,Body#456] > Batched: false, DataFilters: [], Format: Avro, Location: InMemoryFileIndex(1 > paths)[abfss://<container>@<storage>.dfs.core..., PartitionFilters: [], > PushedFilters: [], ReadSchema: > struct<SequenceNumber:bigint,Offset:string,EnqueuedTimeUtc:string,SystemProperties:map<string,str... > {code} -- This message was sent by Atlassian Jira (v8.20.10#820010) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org