Does someone have examples of using the UPathIOManager I che

Question

Does someone have examples of using the UPathIOManager I checked the docs and it s not that obvious to me atm

jamie · Answer

The UPathIOManager is a pretty recent community contribution < Daniel Gafni> i believe it was your PR right do you have some examples of using it

Daniel Gafni · Answer

Sure I have some examples but they don t differ much from the docs because you need to write so little code to use `UPathIOManager` is the example from the docs Here are my examples ```class JSONIOManager UPathIOManager extension str = json def dump to path self context OutputContext obj Any path UPath with path open wb as file file write orjson dumps obj def load from path self context InputContext path UPath gt Any with path open rb as file return orjson loads file read class ParquetIOManager UPathIOManager extension str = pq def dump to path self context OutputContext obj pl DataFrame path UPath with path open wb as file obj write parquet file def load from path self path UPath context InputContext gt pl DataFrame columns = context metadata get columns if columns is not None context log debug f Loading columns= with path open rb as file return pl read parquet file columns=columns def get metadata self context OutputContext obj pl DataFrame gt dict str MetadataValue return report MetadataValue json polars df report obj row count len obj head MetadataValue json orjson loads obj head write json ``` Example for the actual `io manager` ``` io manager config schema= base path Field str is required=False def local parquet io manager init context InitResourceContext gt ParquetIOManager assert init context instance is not None base path = UPath init context resource config get base path init context instance storage directory return ParquetIOManager base path=base path ``` Usage with `S3` would be the same you would just have to define the `AWS ACCESS KEY ID` and `AWS SECRET ACCESS KEY` environment variables when creating the IOManager instance You can specify them as required resources for your `s3 parquet io manager` They will be recognized by `UPath` if the `base path` starts with `s3 ` P S overriding `UPathIOManager get metadata` is a little broken right now failing for loading single inputs but PR should fix it Hopefully the 1 1 8 release will have this merged

Martin Picard · Answer

Hey Daniel thanks a lot for replying I m using Azure so I ended up using azure data lake file system which worked perfectly I was stuck with an issue with the path but I realized I had to cast the UPath to a string and it works beautifully I am using pyarrow parquet so next step will be to figure out how partitioning works with UPathIOManager

Daniel Gafni · Answer

ok cool gt I had to cast the UPath to a string Do you mind elaborating on this issue was it on Dagster s side or on yours gt figure out how partitioning works 1 Loading single partitions works out of the box 2 Loading multiple partitions with `PartitionMapping` also works out of the box UPathIOManager writes asset partitions in the same directory like this ```asset key 2022 01 01 parquet assety key 20 22 01 02 parquet``` I don t think this plays well with `pyarrow` out of the box < sandy> what do you think of extending `UPathIOManager` s functionality by allowing the end user to specify how to format the partition path So something like this would be possible ``` in my awesome IOManager def format partition path context obj pl DataFrame gt str dates in df = obj date unique to list assert len dates in df == 1 assert dates in df 0 == context partition key need to actually parse the date here return f date= context partition key ``` This would result in partitioning structure compatible with the hive schema and pyarrow or Apache Spark would understand it out of the box ```asset key date=2022 01 01 parquet assety key date=20 22 01 02 parquet``` Using empty path if the partitioning is handled inside PySpark and not by Dagster ```def format partition path context obj pl DataFrame gt str return ``` Of course the default formatting would not change

Martin Picard · Answer

```from adlfs import AzureBlobFileSystem from azure identity aio import DefaultAzureCredential from dagster import Any InputContext OutputContext UPathIOManager from pyarrow import fs from upath import UPath from data pipeline import parquet def parquet io manager context Persistent IO manager for PyArrow Parquet account name = context resource config storage account credential = DefaultAzureCredential exclude environment credential=True file system = AzureBlobFileSystem account name=account name credential=credential base path = UPath context resource config container name return ParquetIOManager base path=base path file system=file system class ParquetIOManager UPathIOManager extension str = parquet def init self base path UPath file system fs FileSystem super init base path self fs = file system def load from path self context InputContext path UPath gt Any return parquet read table path filesystem=self fs def dump to path self context OutputContext obj Any path UPath parquet write table obj str path filesystem=self fs ``` the last line was where I was mistakenly using UPath but pyarrow parquet write table wants str or path like object which is confusing because I thought UPath was already path like but pyarrow meant something different haha

Daniel Gafni · Answer

I see so you had to define your own fs object My original idea was to make theUPathIOManagers work with the pure UPaths since the fs can already be supplied inside the user provided base path

Martin Picard · Answer

Doesn t work with azure unfortunately I need to pass azure credentials to the fs

Martin Picard · Answer

I am experiencing weird issues however

Martin Picard · Answer

I think I know the problem but I am not sure how to fix it here is the stacktrace ```The above exception was caused by the following exception PermissionError Errno 13 Permission denied container name File opt data pipeline venv lib python3 9 site packages dagster core execution plan utils py line 56 in op execution error boundary yield File opt data pipeline venv lib python3 9 site packages dagster utils init py line 458 in iterate with context next output = next iterator File opt data pipeline venv lib python3 9 site packages dagster core execution plan execute step py line 626 in gen fn gen output = output manager handle output output context output value File opt data pipeline venv lib python3 9 site packages dagster core storage upath io manager py line 223 in handle output path parent mkdir parents=True exist ok=True File usr local lib python3 9 pathlib py line 1323 in mkdir self accessor mkdir self mode ```

Martin Picard · Answer

the above trace it s running in k8s and it s trying to create a file on the local fs

Martin Picard · Answer

I see some weird behaviour locally as well it creates the container name directory in my local fs then pushes the parquet file to azure blob confused dog

Martin Picard · Answer

I tried different path combinations for example `base path = UPath f abfss context resource config container name ` but I got ```ValueError unable to connect to account for Must provide either a connection string or account name with credentials File Users martin workspace experiment data pipeline venv lib python3 9 site packages dagster core execution plan utils py line 56 in op execution error boundary yield File Users martin workspace experiment data pipeline venv lib python3 9 site packages dagster utils init py line 458 in iterate with context next output = next iterator File Users martin workspace experiment data pipeline venv lib python3 9 site packages dagster core execution plan execute step py line 626 in gen fn gen output = output manager handle output output context output value File Users martin workspace experiment data pipeline venv lib python3 9 site packages dagster core storage upath io manager py line 229 in handle output path parent mkdir parents=True exist ok=True File Users martin workspace experiment data pipeline venv lib python3 9 site packages upath core py line 403 in mkdir self accessor mkdir File Users martin workspace experiment data pipeline venv lib python3 9 site packages upath core py line 134 in getattr self accessor = accessor = self default accessor File Users martin workspace experiment data pipeline venv lib python3 9 site packages upath core py line 28 in init self fs = cls url kwargs File Users martin workspace experiment data pipeline venv lib python3 9 site packages fsspec spec py line 76 in call obj = super call args kwargs File Users martin workspace experiment data pipeline venv lib python3 9 site packages adlfs spec py line 268 in init self do connect File Users martin workspace experiment data pipeline venv lib python3 9 site packages adlfs spec py line 444 in do connect raise ValueError f unable to connect to account for e ``` so I am not sure how if I will get UPathIOManager to work with ablfs

Daniel Gafni · Answer

So that s the thing you need to pass credentials to the base path when creating it This can be done either with kwargs or with environment variables Check out the example notebook and adlfl docs Edit sorry I completely missed that you are passing file system to UPath however I m not near my PC right now and not sure if that s a valid way of creating the UPath

Daniel Gafni · Answer

Re local filesystem in container This has to do with your docker image setup most likely you are running your code in the container under a non root user and didn t chown this directory for this user

Martin Picard · Answer

Yea i never wanted to touch the local filesystem so it s probably the wrong UPath with no protocol doing that right

Martin Picard · Answer

Yea I think in this edge case I won t be able to use the UPathIOManager because I need to pass `storage options= account name ACCOUNT NAME anon False ` to use the `DefaultAzureCredential` because I don t can t use the typical envvars to authenticate I m using `oauth`

Martin Picard · Answer

Thanks for your help thought

Daniel Gafni · Answer

Why can t you use `oauth` in the ` io manager` code and pass the credentials as `storage options` kwargs to `UPath` Something like this ```base path = UPath <abfs my path> account name= account anon=False account key= password ```

Martin Picard · Answer

Woah cool I did not know that I could pass additional arguments other than the 1st arg for path to UPath thanks a lot Daniel

Daniel Gafni · Answer

great does it work we probably should add this to Dagster s docs

Martin Picard · Answer

I used this `base path = UPath f abfss container name account name=account name anon=False ` and it doesn t work I get a please specify container name error

Martin Picard · Answer

comes from dagster core storage upath io manager py line 223

Daniel Gafni · Answer

<https github com dagster io dagster blob master python modules dagster dagster core storage upath io manager py L223|this> line

Daniel Gafni · Answer

Could you post the error

Martin Picard · Answer

Martin Picard · Answer

```The above exception was caused by the following exception ValueError Failed to fetch container properties for for Please specify a container name File Users martin workspace data pipeline venv lib python3 9 site packages dagster core execution plan utils py line 56 in op execution error boundary yield File Users martin workspace data pipeline venv lib python3 9 site packages dagster utils init py line 458 in iterate with context next output = next iterator File Users martin workspace data pipeline venv lib python3 9 site packages dagster core execution plan execute step py line 626 in gen fn gen output = output manager handle output output context output value File Users martin workspace data pipeline venv lib python3 9 site packages dagster core storage upath io manager py line 224 in handle output path parent mkdir parents=True exist ok=True File Users martin workspace data pipeline venv lib python3 9 site packages upath core py line 403 in mkdir self accessor mkdir File Users martin workspace data pipeline venv lib python3 9 site packages upath core py line 67 in mkdir return self fs mkdir File Users martin workspace data pipeline venv lib python3 9 site packages fsspec asyn py line 113 in wrapper return sync self loop func args kwargs File Users martin workspace data pipeline venv lib python3 9 site packages fsspec asyn py line 98 in sync raise return result File Users martin workspace data pipeline venv lib python3 9 site packages fsspec asyn py line 53 in runner result 0 = await coro File Users martin workspace data pipeline venv lib python3 9 site packages adlfs spec py line 1109 in mkdir container exists = await self container exists container name File Users martin workspace data pipeline venv lib python3 9 site packages adlfs spec py line 1084 in container exists raise ValueError The above exception was caused by the following exception ValueError Please specify a container name File Users martin workspace data pipeline venv lib python3 9 site packages adlfs spec py line 1077 in container exists async with self service client get container client File Users martin workspace data pipeline venv lib python3 9 site packages azure storage blob aio blob service client async py line 631 in get container client return ContainerClient File Users martin workspace data pipeline venv lib python3 9 site packages azure storage blob aio container client async py line 120 in init super ContainerClient self init File Users martin workspace data pipeline venv lib python3 9 site packages azure storage blob container client py line 154 in init raise ValueError Please specify a container name ```

Martin Picard · Answer

path parent looks valid

Martin Picard · Answer

and so does path

Martin Picard · Answer

path parent is <abfss dagster> which is a container i have already in the storage account

Daniel Gafni · Answer

can you try specifying a subdirectory for the `base path`

Martin Picard · Answer

if i add test to the end i get another weird error ```pyarrow lib ArrowInvalid Unrecognized filesystem type in URI <abfss dagster test birthdays parquet> File Users martin workspace data pipeline venv lib python3 9 site packages dagster core execution plan utils py line 56 in op execution error boundary yield File Users martin workspace data pipeline venv lib python3 9 site packages dagster utils init py line 458 in iterate with context next output = next iterator File Users martin workspace data pipeline venv lib python3 9 site packages dagster core execution plan execute step py line 626 in gen fn gen output = output manager handle output output context output value File Users martin workspace data pipeline venv lib python3 9 site packages dagster core storage upath io manager py line 226 in handle output self dump to path context=context obj=obj path=path File Users martin workspace data pipeline data pipeline parquet io managers py line 58 in dump to path parquet write table obj str path File Users martin workspace data pipeline venv lib python3 9 site packages pyarrow parquet core py line 2964 in write table with ParquetWriter File Users martin workspace data pipeline venv lib python3 9 site packages pyarrow parquet core py line 948 in init filesystem path = resolve filesystem and path File Users martin workspace data pipeline venv lib python3 9 site packages pyarrow fs py line 185 in resolve filesystem and path filesystem path = FileSystem from uri path File pyarrow fs pyx line 470 in pyarrow fs FileSystem from uri File pyarrow error pxi line 144 in pyarrow lib pyarrow internal check status File pyarrow error pxi line 100 in pyarrow lib check status```

Daniel Gafni · Answer

you don t pass the `UPath s` file to pyarrow You pass the string

Daniel Gafni · Answer

You need to do something like ```with path open wb as file parquet write table obj file ```

Martin Picard · Answer

Still doesn t work

Martin Picard · Answer

``` io manager config schema= container name Field StringSource storage account Field StringSource def parquet io manager context Persistent IO manager for PyArrow Parquet container name = context resource config container name storage account = context resource config storage account credential = DefaultAzureCredential exclude environment credential=True base path = UPath f abfss container name account name=storage account credential=credential return ParquetIOManager base path=base path class ParquetIOManager UPathIOManager extension str = parquet def load from path self context InputContext path UPath gt Any columns = context metadata get columns if columns is not None context log debug f Loading columns= with path open rb as file return parquet read table file columns=columns def dump to path self context OutputContext obj Any path UPath with path open wb as file parquet write table obj file ```

Martin Picard · Answer

```dagster core errors DagsterExecutionHandleOutputError Error occurred while handling output result of step birthdays File Users martin workspace data pipeline venv lib python3 9 site packages dagster core execution plan execute plan py line 266 in dagster event sequence for step for step event in check generator step events File Users martin workspace data pipeline venv lib python3 9 site packages dagster core execution plan execute step py line 394 in core dagster event sequence for step for evt in type check and store output step context user event input lineage File Users martin workspace data pipeline venv lib python3 9 site packages dagster core execution plan execute step py line 451 in type check and store output for evt in store output step context step output handle output input lineage File Users martin workspace data pipeline venv lib python3 9 site packages dagster core execution plan execute step py line 636 in store output for elt in iterate with context File Users martin workspace data pipeline venv lib python3 9 site packages dagster utils init py line 460 in iterate with context return File opt homebrew Cellar python 3 9 3 9 16 Frameworks Python framework Versions 3 9 lib python3 9 contextlib py line 137 in exit self gen throw typ value traceback File Users martin workspace data pipeline venv lib python3 9 site packages dagster core execution plan utils py line 86 in op execution error boundary raise error cls The above exception was caused by the following exception ValueError Failed to fetch container properties for for Please specify a container name File Users martin workspace data pipeline venv lib python3 9 site packages dagster core execution plan utils py line 56 in op execution error boundary yield File Users martin workspace data pipeline venv lib python3 9 site packages dagster utils init py line 458 in iterate with context next output = next iterator File Users martin workspace data pipeline venv lib python3 9 site packages dagster core execution plan execute step py line 626 in gen fn gen output = output manager handle output output context output value File Users martin workspace data pipeline venv lib python3 9 site packages dagster core storage upath io manager py line 223 in handle output path parent mkdir parents=True exist ok=True File Users martin workspace data pipeline venv lib python3 9 site packages upath core py line 403 in mkdir self accessor mkdir File Users martin workspace data pipeline venv lib python3 9 site packages upath core py line 67 in mkdir return self fs mkdir File Users martin workspace data pipeline venv lib python3 9 site packages fsspec asyn py line 113 in wrapper return sync self loop func args kwargs File Users martin workspace data pipeline venv lib python3 9 site packages fsspec asyn py line 98 in sync raise return result File Users martin workspace data pipeline venv lib python3 9 site packages fsspec asyn py line 53 in runner result 0 = await coro File Users martin workspace data pipeline venv lib python3 9 site packages adlfs spec py line 1109 in mkdir container exists = await self container exists container name File Users martin workspace data pipeline venv lib python3 9 site packages adlfs spec py line 1084 in container exists raise ValueError The above exception was caused by the following exception ValueError Please specify a container name File Users martin workspace data pipeline venv lib python3 9 site packages adlfs spec py line 1077 in container exists async with self service client get container client File Users martin workspace data pipeline venv lib python3 9 site packages azure storage blob aio blob service client async py line 631 in get container client return ContainerClient File Users martin workspace data pipeline venv lib python3 9 site packages azure storage blob aio container client async py line 120 in init super ContainerClient self init File Users martin workspace data pipeline venv lib python3 9 site packages azure storage blob container client py line 154 in init raise ValueError Please specify a container name ```

Daniel Gafni · Answer

```base path = UPath f abfss container name account name=storage account credential=credential ``` you removed the subdir again

Daniel Gafni · Answer

what happens here is we are tying to create the parent directory but it can t exist as the container name gets removed Seems like `path parent` should fail in this case Not sure if this is fsspec s bug Anyway just specify any subdir for the base path

Martin Picard · Answer

ah I see

Martin Picard · Answer

so for me container name could be anything as i don t really see it change anything the subdir ends up being the real container name in the azure portal

Martin Picard · Answer

weird

Martin Picard · Answer

I guess I will put `f abfss storage account container name `

Martin Picard · Answer

thanks for helping me debug

Daniel Gafni · Answer

you are welcome hopefully `UPathIOManager` will be convenient to use

Daniel Gafni · Answer

hey < Martin Picard> `hive` formatting can be implemented with `UPathIOManager` by simply using for example `year= Y month= m day= d` for `DailyPartitionsDefinition` s `fmt` Can you try it out