क्या मेटाफ्लो आर्टिफैक्ट का पूर्ण एस 3 यूआरएल पथ प्राप्त करने का कोई तरीका है, जिसे एक चरण में संग्रहीत किया गया था?

मैंने मेटाफ्लो के डेटाआर्टिफैक्ट क्लास को देखा लेकिन एक स्पष्ट एस 3 पथ संपत्ति नहीं देखी।

1
morfys 11 जुलाई 2020, 18:53

2 जवाब

सबसे बढ़िया उत्तर

हाँ, आप कर सकते हैं

Flow('MyFlow')[42]['foo'].task.artifacts.bar._object['location']

जहां MyFlow आपके प्रवाह का नाम है, 42 रन आईडी है, foo विचाराधीन चरण है और bar उस चरण का आर्टिफैक्ट है।

2
Savin 25 फरवरी 2021, 04:42

@ सविन के उत्तर के आधार पर, मैंने एक रन आईडी और आर्टिफैक्ट का नाम दिए गए आर्टिफैक्ट के एस 3 यूआरएल को प्राप्त करने के लिए एक सहायक कार्य लिखा है:

from metaflow import Flow, Metaflow, Run
from typing import List, Union


def get_artifact_s3url_from_run(
    run: Union[str, Run], name: str, legacy_names: List[str] = [], missing_ok: bool = False
) -> str:
    """
    Given a MetaFlow Run and a key, scans the run's tasks and returns the artifact's S3 URL with that key.

    NOTE: use get_artifact_from_run() if you want the artifact itself, not the S3 URL to the artifact.

    This allows us to find data artifacts even in flows that did not finish. If we change the name of an artifact,
    we can support backwards compatibility by also passing in the legacy keys. Note: we can avoid this by resuming a
    specific run and adding a node which re-maps the artifact to another key. This will assign the run a new ID.

    Args:
        missing_ok: whether to allow an artifact to be missing
        name: name of the attribute to look for in task.data
        run: a metaflow.Run() object, or a run ID
        legacy_names: backup names to check

    Returns:
       the value of the attribute. If attribute is not found

    Raises:
        DataartifactNotFoundError if artifact is not found and missing_ok=False
        ValueError if Flow not found
        ValueError if Flow is found but run ID is not.
    """
    namespace(None)  # allows us to access all runs in all namespaces
    names_to_check = [name] + legacy_names
    if isinstance(run, str):
        try:
            run = Run(run)
        except Exception as e:
            # run ID not found. see if we can find other runs and list them
            flow = run.split(sep="/")[0]
            try:
                flow = Flow(flow)
                raise ValueError(f"Could not find run ID {run}. Possible values: {flow.runs()}") from e
            except Exception as e2:
                raise ValueError(f"Could not find flow {flow}. Available flows: {Metaflow().flows}") from e2
    for name_ in names_to_check:
        for step_ in run:
            for task in step_:
                print(f"task {task} artifacts: {task.artifacts} \n \n")
                if task.artifacts is not None and name_ in task.artifacts:
                    # https://stackoverflow.com/a/66361249/4212158
                    return getattr(task.artifacts, name_)._object["location"]

    if not missing_ok:
        raise DataArtifactNotFoundError(
            f"No data artifact with name {name} found in {run}. Also checked legacy names: {legacy_names}"
        )
1
crypdick 29 जून 2021, 21:02