[ https://issues.apache.org/jira/browse/ARROW-10517?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17235388#comment-17235388 ]
Lance Dacey commented on ARROW-10517: ------------------------------------- Latest adlfs (0.5.5): This really creates the test.parquet file as well, not just the directory: {code:java} fs.mkdir("dev/test99999999999/2020/01/28/test.parquet", create_parents=True) {code} And if I try to run the same line again it it fails because the partition exists: {code:python} --------------------------------------------------------------------------- StorageErrorException: Operation returned an invalid status 'The specified blob already exists.' During handling of the above exception, another exception occurred: ResourceExistsError Traceback (most recent call last) /c/airflow/test.py in <module> ----> 6 fs.mkdir("dev/test99999999999/2020/01/28/test.parquet", create_parents=True) ~/miniconda3/envs/airflow/lib/python3.8/site-packages/adlfs/spec.py in mkdir(self, path, delimiter, exist_ok, **kwargs) 880 881 def mkdir(self, path, delimiter="/", exist_ok=False, **kwargs): --> 882 maybe_sync(self._mkdir, self, path, delimiter, exist_ok) 883 884 async def _mkdir(self, path, delimiter="/", exist_ok=False, **kwargs): ~/miniconda3/envs/airflow/lib/python3.8/site-packages/fsspec/asyn.py in maybe_sync(func, self, *args, **kwargs) 98 if inspect.iscoroutinefunction(func): 99 # run the awaitable on the loop --> 100 return sync(loop, func, *args, **kwargs) 101 else: 102 # just call the blocking function ~/miniconda3/envs/airflow/lib/python3.8/site-packages/fsspec/asyn.py in sync(loop, func, callback_timeout, *args, **kwargs) 69 if error[0]: 70 typ, exc, tb = error[0] ---> 71 raise exc.with_traceback(tb) 72 else: 73 return result[0] ~/miniconda3/envs/airflow/lib/python3.8/site-packages/fsspec/asyn.py in f() 53 if callback_timeout is not None: 54 future = asyncio.wait_for(future, callback_timeout) ---> 55 result[0] = await future 56 except Exception: 57 error[0] = sys.exc_info() ~/miniconda3/envs/airflow/lib/python3.8/site-packages/adlfs/spec.py in _mkdir(self, path, delimiter, exist_ok, **kwargs) 918 container=container_name 919 ) --> 920 await container_client.upload_blob(name=path, data="") 921 else: 922 ## everything else ~/miniconda3/envs/airflow/lib/python3.8/site-packages/azure/core/tracing/decorator_async.py in wrapper_use_tracer(*args, **kwargs) 72 span_impl_type = settings.tracing_implementation() 73 if span_impl_type is None: ---> 74 return await func(*args, **kwargs) 75 76 # Merge span is parameter is set, but only if no explicit parent are passed ~/miniconda3/envs/airflow/lib/python3.8/site-packages/azure/storage/blob/aio/_container_client_async.py in upload_blob(self, name, data, blob_type, length, metadata, **kwargs) 715 timeout = kwargs.pop('timeout', None) 716 encoding = kwargs.pop('encoding', 'UTF-8') --> 717 await blob.upload_blob( 718 data, 719 blob_type=blob_type, ~/miniconda3/envs/airflow/lib/python3.8/site-packages/azure/core/tracing/decorator_async.py in wrapper_use_tracer(*args, **kwargs) 72 span_impl_type = settings.tracing_implementation() 73 if span_impl_type is None: ---> 74 return await func(*args, **kwargs) 75 76 # Merge span is parameter is set, but only if no explicit parent are passed ~/miniconda3/envs/airflow/lib/python3.8/site-packages/azure/storage/blob/aio/_blob_client_async.py in upload_blob(self, data, blob_type, length, metadata, **kwargs) 267 **kwargs) 268 if blob_type == BlobType.BlockBlob: --> 269 return await upload_block_blob(**options) 270 if blob_type == BlobType.PageBlob: 271 return await upload_page_blob(**options) ~/miniconda3/envs/airflow/lib/python3.8/site-packages/azure/storage/blob/aio/_upload_helpers.py in upload_block_blob(client, data, stream, length, overwrite, headers, validate_content, max_concurrency, blob_settings, encryption_options, **kwargs) 131 except StorageErrorException as error: 132 try: --> 133 process_storage_error(error) 134 except ResourceModifiedError as mod_error: 135 if not overwrite: ~/miniconda3/envs/airflow/lib/python3.8/site-packages/azure/storage/blob/_shared/response_handlers.py in process_storage_error(storage_error) 145 error.error_code = error_code 146 error.additional_info = additional_data --> 147 raise error 148 149 ResourceExistsError: The specified blob already exists. RequestId:85acda2e-401e-0080-5166-be4d32000000 Time:2020-11-19T11:23:28.7193393Z ErrorCode:BlobAlreadyExists Error:None {code} If I switch to adlfs 0.2.5 (old version which works for ds.dataset()), there error is different when I try to create a directory which already exists but I also cannot create any directory at all for some reason. I also tried to create an entirely new directory which definitely does not exist and ran into an error: {code:python} fs.mkdir("dev/testab1234123/2020/01/28/new.parquet", create_parents=True) RuntimeError: Cannot create dev/testab1234123/2020/01/28/new.parquet. --------------------------------------------------------------------------- RuntimeError Traceback (most recent call last) ----> 6 fs.mkdir("dev/test99999999999/2020/01/28/test.parquet", create_parents=True) ~/miniconda3/envs/old/lib/python3.8/site-packages/adlfs/core.py in mkdir(self, path, delimiter, exists_ok, **kwargs) 561 else: 562 ## everything else --> 563 raise RuntimeError(f"Cannot create {container_name}{delimiter}{path}.") 564 else: 565 if container_name in self.ls("") and path: RuntimeError: Cannot create dev/test99999999999/2020/01/28/test.parquet. {code} But I *am* able to read a dataset which I could *not* do with adlfs 0.5.5 (I get that error about a list of files instead of a dictionary using fs.find() with the latest version). So this is bizarre. I can only read data (with ds.dataset()) with an old version of adlfs, and I can only write data with the newest version. Even pq.read_table() will not work for me using the latest version of adlfs (0.5.5): {code:python} ----> 7 table = pq.read_table(source="dev/testing10/evaluations", columns=None, filters=[('year', '==', '2020')], filesystem=fs) ~/miniconda3/envs/airflow/lib/python3.8/site-packages/pyarrow/parquet.py in read_table(source, columns, use_threads, metadata, use_pandas_metadata, memory_map, read_dictionary, filesystem, filters, buffer_size, partitioning, use_legacy_dataset, ignore_prefixes) 1605 ) 1606 try: -> 1607 dataset = _ParquetDatasetV2( 1608 source, 1609 filesystem=filesystem, ~/miniconda3/envs/airflow/lib/python3.8/site-packages/pyarrow/parquet.py in __init__(self, path_or_paths, filesystem, filters, partitioning, read_dictionary, buffer_size, memory_map, ignore_prefixes, **kwargs) 1465 infer_dictionary=True) 1466 -> 1467 self._dataset = ds.dataset(path_or_paths, filesystem=filesystem, 1468 format=parquet_format, 1469 partitioning=partitioning, ~/miniconda3/envs/airflow/lib/python3.8/site-packages/pyarrow/dataset.py in dataset(source, schema, format, filesystem, partitioning, partition_base_dir, exclude_invalid_files, ignore_prefixes) 669 # TODO(kszucs): support InMemoryDataset for a table input 670 if _is_path_like(source): --> 671 return _filesystem_dataset(source, **kwargs) 672 elif isinstance(source, (tuple, list)): 673 if all(_is_path_like(elem) for elem in source): ~/miniconda3/envs/airflow/lib/python3.8/site-packages/pyarrow/dataset.py in _filesystem_dataset(source, schema, filesystem, partitioning, format, partition_base_dir, exclude_invalid_files, selector_ignore_prefixes) 426 fs, paths_or_selector = _ensure_multiple_sources(source, filesystem) 427 else: --> 428 fs, paths_or_selector = _ensure_single_source(source, filesystem) 429 430 options = FileSystemFactoryOptions( ~/miniconda3/envs/airflow/lib/python3.8/site-packages/pyarrow/dataset.py in _ensure_single_source(path, filesystem) 402 paths_or_selector = [path] 403 else: --> 404 raise FileNotFoundError(path) 405 406 return filesystem, paths_or_selector FileNotFoundError: dev/testing10/evaluations {code} If I turn on use_legacy_dataset=True, then it works though and I am able to use the write_dataset feature. So this is definitely some interaction between the new dataset module and adlfs. {code:python} table = pq.read_table(source="dev/testing10/evaluations", columns=None, filters=[('year', '==', '2020')], filesystem=fs, use_legacy_dataset=True) ds.write_dataset(table, base_dir="dev/adlfs-0.5.5", format="parquet", partitioning=ds.DirectoryPartitioning(pa.schema([("year", pa.int64()), ("month", pa.string()), ("day", pa.string())])), schema=table.schema, filesystem=fs) {code} This does seem to create empty files for each partition as well, which is strange, and the files are named with a part- prefix now instead of a UUID. dev/adlfs-0.5.5/2020/11/15/part-2.parquet dev/adlfs-0.5.5/2020/11/16/part-3.parquet !ss2.PNG! > [Python] Unable to read/write Parquet datasets with fsspec on Azure Blob > ------------------------------------------------------------------------ > > Key: ARROW-10517 > URL: https://issues.apache.org/jira/browse/ARROW-10517 > Project: Apache Arrow > Issue Type: Bug > Components: Python > Affects Versions: 2.0.0 > Environment: Ubuntu 18.04 > Reporter: Lance Dacey > Priority: Major > Labels: azureblob, dataset, dataset-parquet-read, > dataset-parquet-write, fsspec > Attachments: ss.PNG, ss2.PNG > > > > {code:python} > # adal==1.2.5 > # adlfs==0.2.5 > # fsspec==0.7.4 > # pandas==1.1.3 > # pyarrow==2.0.0 > # azure-storage-blob==2.1.0 > # azure-storage-common==2.1.0 > import pyarrow.dataset as ds > import fsspec > from pyarrow.dataset import DirectoryPartitioning > fs = fsspec.filesystem(protocol='abfs', > account_name=base.login, > account_key=base.password) > ds.write_dataset(data=table, > base_dir="dev/test7", > basename_template=None, > format="parquet", > partitioning=DirectoryPartitioning(pa.schema([("year", > pa.string()), ("month", pa.string()), ("day", pa.string())])), > schema=table.schema, > filesystem=fs, > ) > {code} > I think this is due to early versions of adlfs having mkdir(). Although I > use write_to_dataset and write_table all of the time, so I am not sure why > this would be an issue. > {code:python} > --------------------------------------------------------------------------- > RuntimeError Traceback (most recent call last) > <ipython-input-40-bb38d83f896e> in <module> > 13 > 14 > ---> 15 ds.write_dataset(data=table, > 16 base_dir="dev/test7", > 17 basename_template=None, > /opt/conda/lib/python3.8/site-packages/pyarrow/dataset.py in > write_dataset(data, base_dir, basename_template, format, partitioning, > schema, filesystem, file_options, use_threads) > 771 filesystem, _ = _ensure_fs(filesystem) > 772 > --> 773 _filesystemdataset_write( > 774 data, base_dir, basename_template, schema, > 775 filesystem, partitioning, file_options, use_threads, > /opt/conda/lib/python3.8/site-packages/pyarrow/_dataset.pyx in > pyarrow._dataset._filesystemdataset_write() > /opt/conda/lib/python3.8/site-packages/pyarrow/_fs.pyx in > pyarrow._fs._cb_create_dir() > /opt/conda/lib/python3.8/site-packages/pyarrow/fs.py in create_dir(self, > path, recursive) > 226 def create_dir(self, path, recursive): > 227 # mkdir also raises FileNotFoundError when base directory is > not found > --> 228 self.fs.mkdir(path, create_parents=recursive) > 229 > 230 def delete_dir(self, path): > /opt/conda/lib/python3.8/site-packages/adlfs/core.py in mkdir(self, path, > delimiter, exists_ok, **kwargs) > 561 else: > 562 ## everything else > --> 563 raise RuntimeError(f"Cannot create > {container_name}{delimiter}{path}.") > 564 else: > 565 if container_name in self.ls("") and path: > RuntimeError: Cannot create dev/test7/2020/01/28. > {code} > > Next, if I try to read a dataset (keep in mind that this works with > read_table and ParquetDataset): > {code:python} > ds.dataset(source="dev/staging/evaluations", > format="parquet", > partitioning="hive", > exclude_invalid_files=False, > filesystem=fs > ) > {code} > > This doesn't seem to respect the filesystem connected to Azure Blob. > {code:python} > --------------------------------------------------------------------------- > FileNotFoundError Traceback (most recent call last) > <ipython-input-41-4de65fe95db7> in <module> > ----> 1 ds.dataset(source="dev/staging/evaluations", > 2 format="parquet", > 3 partitioning="hive", > 4 exclude_invalid_files=False, > 5 filesystem=fs > /opt/conda/lib/python3.8/site-packages/pyarrow/dataset.py in dataset(source, > schema, format, filesystem, partitioning, partition_base_dir, > exclude_invalid_files, ignore_prefixes) > 669 # TODO(kszucs): support InMemoryDataset for a table input > 670 if _is_path_like(source): > --> 671 return _filesystem_dataset(source, **kwargs) > 672 elif isinstance(source, (tuple, list)): > 673 if all(_is_path_like(elem) for elem in source): > /opt/conda/lib/python3.8/site-packages/pyarrow/dataset.py in > _filesystem_dataset(source, schema, filesystem, partitioning, format, > partition_base_dir, exclude_invalid_files, selector_ignore_prefixes) > 426 fs, paths_or_selector = _ensure_multiple_sources(source, > filesystem) > 427 else: > --> 428 fs, paths_or_selector = _ensure_single_source(source, > filesystem) > 429 > 430 options = FileSystemFactoryOptions( > /opt/conda/lib/python3.8/site-packages/pyarrow/dataset.py in > _ensure_single_source(path, filesystem) > 402 paths_or_selector = [path] > 403 else: > --> 404 raise FileNotFoundError(path) > 405 > 406 return filesystem, paths_or_selector > FileNotFoundError: dev/staging/evaluations > {code} > This *does* work though when I list the blobs before passing them to > ds.dataset: > {code:python} > blobs = wasb.list_blobs(container_name="dev", prefix="staging/evaluations") > dataset = ds.dataset(source=["dev/" + blob.name for blob in blobs], > format="parquet", > partitioning="hive", > exclude_invalid_files=False, > filesystem=fs) > {code} > Next, if I downgrade to pyarrow 1.0.1, I am able to read datasets (but there > is no write_datasets): > {code:python} > # adal==1.2.5 > # adlfs==0.2.5 > # azure-storage-blob==2.1.0 > # azure-storage-common==2.1.0 > # fsspec==0.7.4 > # pandas==1.1.3 > # pyarrow==1.0.1 > dataset = ds.dataset("dev/staging/evaluations", format="parquet", > filesystem=fs) > dataset.to_table().to_pandas() > {code} > edit: > pyarrow 2.0.0 > fsspec 0.8.4 > adlfs v0.5.5 > pandas 1.1.4 > numpy 1.19.4 > azure.storage.blob 12.6.0 > {code:python} > x = adlfs.AzureBlobFileSystem(account_name=name, account_key=key) > type(x.find("dev/test", detail=True)) > list > fs = fsspec.filesystem(protocol="abfs", account_name=name, account_key=key) > type(fs.find("dev/test", detail=True)) > list > {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)