Remove cloud specific prefix #37
Conversation
| # existing_pipelines = client.list_pipelines(page_size=100).pipelines | ||
| # for existing_pipeline in existing_pipelines: | ||
| # if existing_pipeline.name == pipeline_name: | ||
| # # Delete existing pipeline before uploading | ||
| # logger.warning( | ||
| # f"Pipeline {pipeline_name} already exists. Deleting old pipeline..." | ||
| # ) | ||
| # client.delete_pipeline_version(existing_pipeline.default_version.id) | ||
| # client.delete_pipeline(existing_pipeline.id) |
There was a problem hiding this comment.
This is a temporary fix in order to compile the pipeline.
| # add subset prefix to columns | ||
| df = df.rename( | ||
| columns={ | ||
| col: name + "_" + col for col in df.columns if col not in index_fields |
There was a problem hiding this comment.
For joining all the data into 1 dataframe we will need to join on the index which gets filtered out here.
There was a problem hiding this comment.
The index is not filtered out, this operation only renames the non-index columns
| if index is None: | ||
| index_df = self._load_index() | ||
| ids = index_df["id"].compute() | ||
| sources = index_df["source"].compute() |
There was a problem hiding this comment.
So there is no workaround for having to bring those in memory before filtering?
There was a problem hiding this comment.
In order to filter a dataframe based on an index, I assume that you need to have the entire index present.
Correct me if I'm wrong @GeorgesLorre @RobbeSneyders
There was a problem hiding this comment.
I would have to look into Dask in more detail before I can give input on this.
Remove cloud specific prefix
This PR removes the need for a specific prefix, as we can just include that in the base path of the configuration.