https://dagster.io/ logo
#ask-community
Title
# ask-community
b

Binoy Shah

02/22/2024, 8:54 PM
I am using
dagster-polars
library for PolarsParquetIOManager. I am struggling to set up correct S3 Access. Our S3 credentials are not in their standard format and i need to copy it from some other place, How do I provide the PolarsParquetIOManager correct credentials cc @Daniel Gafni
d

Daniel Gafni

02/22/2024, 8:56 PM
Hey! You should be able to provide
storage_options
like this:
Copy code
PolarsParquetIOManager(..., cloud_storage_options={...})
This feature is a recent addition. I think @Ion Koutsouris has this working in production. Let me know if this works for you
b

Binoy Shah

02/22/2024, 8:57 PM
Cool, thanks, trying it now
d

Daniel Gafni

02/22/2024, 8:57 PM
I think both
storage_options
and
cloud_storage_options
should work.
storage_options
is a Pydantic alias to
cloud_storage_options
, which is the original field.
storage_options
is an alias available for configuration from dicts/yaml
👍 1
Code for reference
Actually seems like aliases only affect serialization/deserialization. You would have to use
cloud_storage_options
when instantiating the class.
b

Binoy Shah

02/22/2024, 9:15 PM
Yup, I was going to update you that info..
fixing it..
d

Daniel Gafni

02/22/2024, 11:12 PM
So does it work?
b

Binoy Shah

02/23/2024, 1:33 PM
some other complications came up.. will surely update you after verification
👍 1
Using the attached code as shown in snippet at the bottom I still am getting errors as below.
Copy code
polars.exceptions.ComputeError: Generic S3 error: Error after 0 retries in 300.000908139s, max_retries:2, retry_timeout:10s, source:error sending request for url (<http://169.254.169.254/latest/api/token>): operation timed out
  File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/utils.py", line 54, in op_execution_error_boundary
    yield
  File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/inputs.py", line 831, in _load_input_with_input_manager
    value = input_manager.load_input(context)
  File "/usr/local/lib/python3.10/site-packages/dagster_polars/io_managers/base.py", line 219, in load_input
    return self._load_single_input(path, context)
  File "/usr/local/lib/python3.10/site-packages/dagster/_core/storage/upath_io_manager.py", line 239, in _load_single_input
    obj = self.load_from_path(context=context, path=path)
  File "/usr/local/lib/python3.10/site-packages/dagster_polars/io_managers/base.py", line 319, in load_from_path
    ldf = self.scan_df_from_path(path=path, context=context)  # type: ignore
  File "/usr/local/lib/python3.10/site-packages/dagster_polars/io_managers/parquet.py", line 276, in scan_df_from_path
    ldf = scan_parquet(path, context)
  File "/usr/local/lib/python3.10/site-packages/dagster_polars/io_managers/parquet.py", line 75, in scan_parquet
    return pl.scan_parquet(str(path), storage_options=storage_options, **kwargs)  # type: ignore
  File "/usr/local/lib/python3.10/site-packages/polars/utils/deprecation.py", line 136, in wrapper
    return function(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/polars/utils/deprecation.py", line 136, in wrapper
    return function(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/polars/io/parquet/functions.py", line 311, in scan_parquet
    return pl.LazyFrame._scan_parquet(
  File "/usr/local/lib/python3.10/site-packages/polars/lazyframe/frame.py", line 464, in _scan_parquet
    self._ldf = PyLazyFrame.new_from_parquet(
@Daniel Gafni edit. wierdness of slack showing badly ordered snippets.
d

Daniel Gafni

02/23/2024, 3:18 PM
I can think of 3 steps: 1. @Ion Koutsouris please let us know if you have this working in prod. Are there any errors above? 2. @Binoy Shah meanwhile, you can try to debug this by overriding this method (
def scan_df_from_path
) and changing this function usage it in. Perhaps you can print the values available in
path.storage_options
. 3. I think I might rework this by directly passing
self.cloud_storage_options
to data reading code instead of relying on
path.storage_options
. Also, can you try writing a parquet?
i

Ion Koutsouris

02/23/2024, 3:23 PM
There were some issues with Polars and S3 not so long ago, something to do with concurrency budget in Tokio, so would be good to share what version you have
b

Binoy Shah

02/23/2024, 3:24 PM
polars = { version = "^0.20.5", extras = ["numpy", "pyarrow", "connectorx" , "fsspec"] }
i

Ion Koutsouris

02/23/2024, 3:24 PM
Ok point 3, you should double check what comes out of UPath
Because we are passing the storage options into UPath and then its upaths responsibility to construct the file system and storage options, if the Uri or storage options are not recognized then they are not properly passed through
d

Daniel Gafni

02/23/2024, 3:26 PM
Yes, I suspect this may be happening here. Try to print these values or jump into a remote debugger.
b

Binoy Shah

02/23/2024, 3:26 PM
in another code I do this
Copy code
data_dictionary_latest: LazyFrame = pl.scan_parquet(source=survey_dict_in_s3, storage_options=_aws_secrets_from_env())
and that works
Copy code
def _aws_secrets_from_env():
    return {
        "aws_access_key_id": env_or_value("ACCESS_KEY"),
        "aws_secret_access_key": env_or_value("SECRET_KEY"),
        "aws_session_token": env_or_value("AWS_SESSION_TOKEN"),
        "aws_region": "us-east-1",
    }
@Ion Koutsouris when you say check what comes out of UPath, can you elaborate ?
i

Ion Koutsouris

02/23/2024, 3:29 PM
Try to manually create UPath object
uPath(uri, **storage_options)
b

Binoy Shah

02/23/2024, 3:30 PM
Oh okay.. checking now
i

Ion Koutsouris

02/23/2024, 3:30 PM
Then jnspect the storage options in the object instance
d

Daniel Gafni

02/23/2024, 3:30 PM
Yeah check what's in
UPath(uri, **storage_options).storage_options
b

Binoy Shah

02/23/2024, 3:40 PM
Copy code
my_up:UPath = UPath("<s3://bucket/dagster-io-parquet/rce_shared/SURVEY_RESPONSE_DATA_DICTIONARY.parquet>", **s3_io_creds )
my_up.storage_options
Out[34]: 
{'aws_access_key_id': 'AS99999999999',
 'aws_secret_access_key': 'DAu999999999St',
 'aws_session_token': 'FwoB53SI'}
So that's appropriate. Its not from dagster runtime, its on my python console ☝️
i

Ion Koutsouris

02/23/2024, 3:50 PM
Try bumping polars
d

Daniel Gafni

02/23/2024, 3:56 PM
I suggest printing values/debugging inside the actual IOManager code just to be sure these values are passed to
pl.scan_parquet
b

Binoy Shah

02/23/2024, 4:01 PM
is there a way I can directly invoke PolarsParquetIOManager to read the parquet file ?
d

Daniel Gafni

02/23/2024, 4:06 PM
Perhaps this code might be the easiest way:
Copy code
from dagster import materialize, asset, SourceAsset

upstream_asset = SourceAsset(["my_bucket", "my_key"], io_manager_def=PolarsParquetIOManager())

@asset
def my_asset(upstream_asset):
    ...

materialize([upstream_asset, my_asset])
b

Binoy Shah

02/23/2024, 4:07 PM
Okay, checking it out
d

Daniel Gafni

02/23/2024, 4:07 PM
you can then insert a breakpoint in the PolarsParquetIOManager
👍 1
b

Binoy Shah

02/23/2024, 9:22 PM
So Some progress @Daniel Gafni, It fails only in Kubernetes, but not in local When I run it in an asset, and print storage Options
print(context.resources.polars_io_manager_v2.storage_options)
it prints out
{}
d

Daniel Gafni

02/23/2024, 10:32 PM
So seems like an issue on your side? If it works locally it should also work in K8s (given the setup is correct)
b

Binoy Shah

02/23/2024, 10:35 PM
That's what I am assuming, but I have Env variables setup a bit differently, for s3 credentials. My Local S3 Credential are in its Provider Chain, so the default aws credential profile has necessary credentials, although I am injecting the variables as I described above in the thread, i suspect it's falling back to its default credential chain
where'as Kubernetes does not have the default credential chain..
~/.aws/credentials
file is not present in kube pod
its only environ var
d

Daniel Gafni

02/23/2024, 10:51 PM
I think the first step should be figuring out how to load the Parauet file with raw polars Should be easy if you can exec in your pod
d

Daniel Gafni

02/23/2024, 10:54 PM
I see. In this case your goal should be seeing the same dictionary in path.storage_options inside the IOManager.
b

Binoy Shah

02/23/2024, 10:55 PM
could it be that I need to do late binding and my dictionary should be the form of
Copy code
aws_access_key_id:
        env: AWS_ACCESS_KEY_ID
      aws_secret_access_key:
        env: AWS_SECRET_ACCESS_KEY
      aws_session_token:
        env: AWS_SESSION_TOKEN
such that the
_process_env_vars
evaluates it late
?
right now the code to build PolarsParquetIOManager https://dagster.slack.com/files/U03AE3Z2A0H/F06LBT91B42/untitled.py is invoked at startup of the App, but as resource, and in dagster pod runs, i might need to do late binding by associating the env keys
d

Daniel Gafni

02/23/2024, 11:12 PM
Can’t look at the code right now but I’m pretty sure cloud_storage_options supposed to be used with EnvVar. You should pass a dict of EnvVars to it. I think my previous example was misleading.
b

Binoy Shah

02/24/2024, 1:25 AM
No worries, thanks for the help and have a great weekend
Hey @Daniel Gafni just to update you, I couldnt make the
storage_options
/
cloud_storage_options
work, not sure if it was my problem or some inherent underlying UPath or other translation issue. I finally remedied it by converting my non-conforming credential fields to AWS recognized ENV variables at the app startup in the
__init__.py
file and it worked.
Copy code
if getenv("IS_CI", "default") == "default":
    environ["AWS_ACCESS_KEY_ID"] = getenv("ACCESS_KEY")
    environ["AWS_SECRET_ACCESS_KEY"] = getenv("SECRET_KEY")
    environ["AWS_SESSION_TOKEN"] = getenv("AWS_SESSION_TOKEN")
    environ["AWS_REGION"] = getenv("AWS_REGION", "us-east-1")
9 Views