apache / apache/beam

[Bug]: WriteToParquet to Azure Blob Storage not working in Spark

Open
#24,303 7 comments 0 reactions 0 assignees View on GitHub
bug io P2 parquet python
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.