Flyte flyte:// file system and improve remote file handling

[wild-endeavor]

TL;DR

Examples of improvements

• pyflyte run used to use a special way of handling certain objects, now that is centralized to use the typeEngine
• pyflyte run and gate node used different string parsing systems, now that is centralized
• pyflyte run can now use multiple files in a directory locally to create a new remote FlyteDirectory representation
• pyflyte run code has been simplified
• workflow defaults that point to local files/folders are auto-uploaded and includes limited support for dataframes. Also data is maintained using hashes so enables versioning

New additions

• pyflyte info added
• pyflyte fetch added
• This introduces new flyte-uri concept that is available in the UI and you can retrieve inputs and outputs from any execution (workflow or even a single node) instantaneously as it deeplinks into it.
  centralizes fsspec as the way to interact with data

Planned work

• Artifacts service that is unblocked and simplified with this
• improved launchform experience from flyte uris
  and many more data interaction improvements

Details

What it does?

• adds a new fsspec file system where flyte:// is the protocol. This is not a real filesystem though, so it's only used in pyflyte run and pyflyte register.
• cleans up code around data persistence.

The reason the file system is useful:

• Defaults for workflows that are of offloaded types can get uploaded properly during pyflyte run/register.
• pyflyte run currently duplicates a bunch of work that's supposed to be handled by the type engine because it needs to handle remote upload. This can be cleaned up in the future if the data proxy service can be accessible at a lower level.

The drawback of this approach:

• The data proxy layer is not a real filesystem. Currently you can't know beforehand where the thing you upload will land. That is, if you think about writing files as a cp src flyte://dest command, the destination is ignored. For this reason, rpath in the implementation is always ignored and a placeholder is used instead. Instead, the remote_fs implementation returns the s3 native url as part of the put call.
• If your default input is a file/folder/pickle, it works, but if it's a dataframe, it works, but sadly, you may end up with duplicate registration problems, the reasons are as follows,

This is basically an fsspec filesystem that writes to a different location than what is specified. However, it's not always possible to capture the output where the file was really written to. More concretely, if you do:

df.to_parquet("s3://a/b/c.parquet")

pandas will use fsspec to fetch the file system, call open on it, and then write to the open buffer. After writing, users expect that s3 location to hold the contents of the dataframe. With the flyte remote fs, if you called df.to_parquet("flyte://anything"), the dataframe contents do not end up there. Instead they end up in some s3:// location (assuming s3 is the blob store).

Because of this, calls to put_data were updated to read back the remote_path, now returned by put_data. (this is why dataframes don't work as default input. we can of course add another transformer that writes to a memory buffer or file first of course only for pyflyte run/register.)

To promote the behavior, a new function was added to data_persistence put_raw_data which writes always to the raw output data prefix. See the function and corresponding test. This PR does not update all the calls yet to the new interface.

The flyte fs also does not support open. The reason is simply that the data proxy call requires that we know the MD5 hash. This doesn't really make sense in the context of an open call where the data might not even be known yet. If this wasn't the case, we could've written the pandas transformer to fs.open() and then write directly to the buffer.

Also,

• The old get_random_remote_path/directory was removed from data persistence and rewritten to be based on an input fs.

Type

• Bug Fix
• Feature
• Plugin

Are all requirements met?

• Code completed
• Smoke tested
• Unit tests added
• Code documentation added
• Any pending items have an associated Issue

Complete description

Smaller changes

• Add length to hash_file
• Added a limited-use (only Remote uses it for now) local fsspec implementation which overrides the / field for proper Windows use.

Other thoughts

We also briefly thought about replacing the fsspec s3 filesystem with something that will do the signed urls but this felt more complicated and felt like it interfered too much with fsspec. Also would be dependent on what blob store admin was configured to use.

Needs: flyteorg/flyteidl#416 and flyteorg/flyteadmin#577
flyteorg/flytestdlib#160 and flyteorg/flyteadmin#587

Tracking Issue

NA

[kumare3]

@pingsutw / @wild-endeavor / @eapolinario this PR is ready and works in cases I have tested. I think a few more unit tests would be good. But, one problem. take the following example

df = pd.DataFrame({"Name": ["Tom", "Joseph11"], "Age": [20, 22]})

@workflow
def wf(sd: StructuredDataset = StructuredDataset(dataframe=df)) -> StructuredDataset:
    return t1(sd=sd)

This currently does not work. With this PR, this will work, but will cause "registration clashes" even for registering the same code again, as the path is not auto de-duped.
As this does not work today, it should not break any old code, but may break new code.

The problem is because we are using pandas parquet writer, we cannot control the remote path. The remote path is created as a random path on the raw prefix. One solution might be to hash the data value, but this is expensive (though completely ok for the registration case). Identifying the difference between registration vs runtime may be tough.

I am still investigating, but want to make the PR ready for review

[pingsutw]

LGTM, some questions in the comment

[pingsutw]

Do we need to use random paths? As in this case

[kumare3]

done

[pingsutw]

❤️

[pingsutw]

do we need to override here? I thought type_engine would modify it

[kumare3]

i think we are going to override everywhere. TypeEngine could modify, but i want to be sure, we should do it wherever we can

[pingsutw]

Suggested change

	# h = hashlib.md5()
	# h.update(data)
	# md5 = h.digest()
	# l = len(data)
	#
	# headers = {"Content-Length": str(l), "Content-MD5": md5}

[Tom-Newton]

Hello. I've been trying to upgrade from flytekit 1.9.1 to 1.10.2 and I think the changes from this PR are causing some issues for me when I try to use a FlyteFile as an input to a workflow that I submit using pyflyte.

I've left a couple of comments but in addition to that I'm having issues when re-running the same workflow with the same FllyteFile.

	details = "file already exists at location [<path>], specify a matching hash if you wish to rewrite"
	debug_error_string = "UNKNOWN:Error received from peer ipv4:172.27.128.127:81 {created_time:"2024-01-22T21:59:07.492837995+00:00", grpc_status:6, grpc_message:"file already exists at location [<path>], specify a matching hash if you wish to rewrite"}"

This one confuses me because it looks like its doing the same as before except previously it didn't fail.

[Tom-Newton]

It seems to me like this should be replacing REMOTE_PLACEHOLDER. We call

 res = await super()._put(lpath, REMOTE_PLACEHOLDER, recursive, callback, batch_size, **kwargs)

So I think the rpath is going to be just REMOTE_PLACEHOLDER. For me this results in paths that look like

abfs://flyte/flyteexamples/development/HAZFZJWIE6KZ3TWFZJJRFAXH6A======/flyte://data

This does work but it doesn't seem quite right to me.

[wild-endeavor]

yeah i ran into this issue too. working on a fix. but this is opt-in behavior isn't it?

[Tom-Newton]

Sorry, I don't understand how its opt-in. pyflyte run depends on this if you use FlyteFile.

[wild-endeavor]

#2121 this is the pr i'm working on, but not sure what's going on with the test failures, will have to take a look later this week.

[Tom-Newton]

I think this is a problem for Azure users. For writes to Azure blob storage the "x-ms-blob-type": "BlockBlob" header is required. Without it we get

│ Invalid value for '--segment_list_file': Failed to convert param: <Option segment_list_file>, value: /home/tomnewton/WayveCode/wayve/services/data/pipelines/flyte/examples/fanout_workflow/run_windows_list.txt to type: <class 'flytekit.types.file.file.FlyteFile'>. Reason 400,      │
│ message="An HTTP header that's mandatory for this request is not specified.",                                                                                                                                                                                                                

In the old code path there was a hack to add in the extra header required for Azure

flytekit/flytekit/remote/remote.py

Line 825 in cba830e

extra_headers = self.get_extra_headers_for_protocol(upload_location.native_url)

#1784

[Tom-Newton]

BTW I worked out the file already exists at location [<path>], specify a matching hash if you wish to rewrite error. It appears this is because remote_fs.get_upload_link sets the filename_root argument when it calls SynchronousFlyteClient.get_upload_signed_url while the old remote uploads in pyflyte did not.