[Bug]: WriteToParquet to Azure Blob Storage not working in Spark
- Dominant language
- Java
- Stars
- 8.7k
- Forks
- 4.7k
- Avg merge
- 1d 20h
- Merged PRs (30d)
- 196
Description
### What happened?
I am running Apache Beam on Spark in Azure. My code:
```py
from apache_beam import Create, Map, Pipeline
from apache_beam.io.parquetio import WriteToParquet
from apache_beam.options.pipeline_options import PipelineOptions
options = PipelineOptions(
[
"--runner", "SparkRunner",
"--spark_version", "3",
]
)
with Pipeline(options=options) as pipeline:
(
pipeline
| Create(list(range(1000)))
| Map(lambda x: {"name": str(x), "age": x})
| WriteToParquet(
"azfs://mystorageaccount/mycontainer/mypath/",
pyarrow.schema([("name", pyarrow.binary()), ("age", pyarrow.int64())]),
file_name_suffix=".parquet",
)
)
```
The error:
```
Traceback (most recent call last):
File "apache_beam/runners/common.py", line 1198, in apache_beam.runners.common.DoFnRunner.process
File "apache_beam/runners/common.py", line 718, in apache_beam.runners.common.PerWindowInvoker.invoke_process
File "apache_beam/runners/common.py", line 841, in apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window
File "apache_beam/runners/common.py", line 1334, in apache_beam.runners.common._OutputProcessor.process_outputs
File "/databricks/python3/lib/python3.7/site-packages/apache_beam/io/iobase.py", line 1229, in
window.TimestampedValue(v, timestamp.MAX_TIMESTAMP) for v in outputs)
File "/databricks/python3/lib/python3.7/site-packages/apache_beam/io/filebasedsink.py", line 279, in finalize_write
self._check_state_for_finalize_write(writer_results, num_shards))
File "/databricks/python3/lib/python3.7/site-packages/apache_beam/io/filebasedsink.py", line 263, in _check_state_for_finalize_write
FileSystems.checksum(src) == FileSystems.checksum(dst)):
File "/databricks/python3/lib/python3.7/site-packages/apache_beam/io/filesystems.py", line 324, in checksum
return filesystem.checksum(path)
File "/databricks/python3/lib/python3.7/site-packages/apache_beam/io/azure/blobstoragefilesystem.py", line 277, in checksum
raise BeamIOError("Checksum operation failed", {path, e})
apache_beam.io.filesystem.BeamIOError: Checksum operation failed with exceptions {'azfs://mystorageaccount/mycontainer/mypath/beam-temp--9a346cf260e611ed84fc00163eafa21b/59821b95-11a8-43fd-9117-f9b8466f46fd..parquet', BlobStorageError('The specified blob does not exist.', 404)}
```
This error is only happening when I have Spark speculative execution enabled. When I disable speculative execution, the problem doesn't happen. It seems to me like `finalize_write` is not idempotent here, and is deleting temp files needed by other invocations.
### Issue Priority
Priority: 2
### Issue Component
Component: io-py-parquet
Contributor guide
Assessment
This issue has not been assessed yet.