New RunLLM response generated >>> Question Hi I m trying to Flyte #runllm-flyte-trial

New RunLLM response generated: >>>*# Question* Hi...

ancient-wolf-19325

07/24/2024, 7:28 PM

New RunLLM response generated: >>>*# Question* Hi, I'm trying to use a combination of dynamic workflows + map_task to perform data processing operations on a list of custom data. The input to the

main()

workflow is a

List[CustomDataclass]

. This list is then batched into smaller lists, creating

List[List[CustomDataclass]]

, and passed to a dynamic workflow

level1_wf

. The

level1_wf

workflow calls

map_task()

to let the

process_data

task operate on

CustomDataclass

. Here's how my code is structured:

Copy code

@task
def process_data(input_data: CustomDataclass) -> Dict[str, str]:
    return processing(input_data)

@dynamic
def map_data_proc_tasks(data_chunks: List[List[BinDataModel]]) -> List[List[Dict[str, str]]:
    chunk_results = []
    for chunk in data_chunks:
        result = map_task(process_data, concurrency=16)(input_data=chunk)
        chunk_results.append(result)
    return chunk_results

@task
def merge_results(results_list: List[List[Dict[str, str]]]) -> JSONLFile:
    merged_data = []
    for result in results_list:
        merged_data.extend(result)
    _, merged_filepath = tempfile.mkstemp(suffix=".json")
    write_json_file(merged_filepath, merged_data)
    return merged_filepath

@workflow
def wf_transform_input_imgs(data_list: List[CustomDataClass]) -> JSONLFile:
    data_chunks = chunk_list(input_list=data_list, chunk_size=64)
    chunk_results = map_data_proc_tasks(data_chunks)
    return merge_results(chunk_results)

When I try to register the tasks and workflows, I get this error:

Copy code

Request rejected by the API, due to Invalid input.
RPC Failed, with Status: StatusCode.INVALID_ARGUMENT
        details: map_data_proc_tasks task with different structure already exists

I've tried a teardown and restart of the sandbox cluter, but that doesn't help. What am I doing wrong here ? # Answer The error you're encountering,

StatusCode.INVALID_ARGUMENT

, indicates that a task with the same name but a different structure already exists in the Flyte registry. This can happen if you have previously registered a task with the same name but with different input/output types or other structural changes. Steps to Resolve the Issue: 1. Update Task Versions: Ensure that you are using a new version for your tasks and workflows when you make structural changes. Flyte uses versions to differentiate between different iterations of the same task or workflow. 2. Check for Structural Changes: Verify that the structure of your tasks and workflows has not changed between registrations. If it has, you need to update the version. 3. Use Unique Names: If you are frequently changing the structure, consider using unique names for your tasks and workflows to avoid conflicts. Example Code with Versioning: Here's how you can modify your code to include versioning:

Copy code

from flytekit import task, dynamic, workflow, map_task
from typing import List, Dict
import tempfile
import json

ancient-wolf-19325

07/24/2024, 7:28 PM

>>```

class CustomDataClass: # Define your custom data class here pass def processing(input_data: CustomDataClass) -> Dict[str, str]: # Your processing logic here return {"key": "value"} @task(version="v1") def process_data(input_data: CustomDataClass) -> Dict[str, str]: return processing(input_data) @dynamic(version="v1") def map_data_proc_tasks(data_chunks: List[List[CustomDataClass]]) -> List[List[Dict[str, str]]]: chunk_results = [] for chunk in data_chunks: result = map_task(process_data, concurrency=16)(input_data=chunk) chunk_results.append(result) return chunk_results @task(version="v1") def merge_results(results_list: List[List[Dict[str, str]]]) -> str: merged_data = [] for result in results_list: merged_data.extend(result) _, merged_filepath = tempfile.mkstemp(suffix=".json") with open(merged_filepath, 'w') as f: json.dump(merged_data, f) return merged_filepath @workflow(version="v1") def wf_transform_input_imgs(data_list: List[CustomDataClass]) -> str: data_chunks = chunk_list(input_list=data_list, chunk_size=64) chunk_results = map_data_proc_tasks(data_chunks=data_chunks) return merge_results(results_list=chunk_results) ``` Additional Considerations: - Sandbox Restart: Restarting the sandbox might not clear the registered tasks and workflows. Ensure that you are using a new version or unique names to avoid conflicts. - FlyteAdmin API: Flyte does not provide a delete API for tasks and workflows to prevent accidental deletions. You can only archive them. If you need to completely remove a task or workflow, you might need to manually delete it from the database, but this is not recommended and should be done with caution. Would you like more details on any specific part of this process? Sources: - Flyte GitHub Issue #2508 - Flyte GitHub Issue #4702 - Flyte GitHub Issue #5473 - Flyte Admin API Documentation Slack Link https://flyte-org.slack.com/archives/CP2HDHKE1/p1721849309.003449 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

Open in Slack

Previous Next