Hello, I am having a hard time to pass optional input param to tasks. Ideally, I want the task either be passed the actual instance of the input param or None. So I have :

from sklearn.preprocessing import OneHotEncoder
@task
def transform(
    encoder: Optional[OneHotEncoder] = None, ...
)

The workflow will do:

return transform(
   encoder=None, ...

This works when running locally. However, if I bake the module containing the task transform() into an image and refer to it in a remote run, no mater what I do, I keep getting

flytekit/core/type_engine.py", line 1060, in to_literal
    raise TypeError("Ambiguous choice of variant for union type")
...
flytekit/core/promise.py", line 937, in create_and_link_node
    raise AssertionError(f"Failed to Bind variable {k} for function {entity.name}.") from e
AssertionError: Failed to Bind variable encoder for function dai_mle_models.onehotenc.simple_onehotenc.transform.

Note, another input param in a different task defined in the same way except it’s of type pandas.DataFrame, works. Why is that?

def estimate(test_data: Optional[pd.DataFrame] = None, ...

Hey all, I believe I found a small bug. LMK if there is a better channel for this. During local execution, conditionals interpret a int variable with value

0

as None.

@workflow
def _impl(
    delta: int,
) -> Set[str]:
    return (
        conditional("terminal_case")
        .if_((delta != -1) & (delta <= 10))
        .then(end())
        .else_()
        .then(
            iterate()
        )
    )


def cluster_distributed() -> Set[str]:
    _impl(delta=0)

Outputs an error comparing NoneType <= 10

Folks the core team at union are currently unavailable- 2 had a baby, 1 got married etc and so we might be a little slow in responding. Community please help the new folks

Hi everyone, I'm following along the tutorial in the Getting Started section. I'm trying to run the sample example.py locally but kept running into this error. Any change someone could help me with this?

File "/Library/Frameworks/Python.framework/Versions/3.11/bin/pyflyte", line 5, in <module>
    from flytekit.clis.sdk_in_container.pyflyte import main
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/flytekit/__init__.py", line 165, in <module>
    from flytekit.core.base_sql_task import SQLTask
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/flytekit/core/base_sql_task.py", line 4, in <module>
    from flytekit.core.base_task import PythonTask, TaskMetadata
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/flytekit/core/base_task.py", line 26, in <module>
    from flytekit.configuration import SerializationSettings
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/flytekit/configuration/__init__.py", line 473, in <module>
    @dataclass(init=True, repr=True, eq=True, frozen=True)
     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/dataclasses.py", line 1210, in wrap
    return _process_class(cls, init, repr, eq, order, unsafe_hash,
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/dataclasses.py", line 958, in _process_class
    cls_fields.append(_get_field(cls, name, type, kw_only))
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/dataclasses.py", line 815, in _get_field
    raise ValueError(f'mutable default {type(f.default)} for field '
ValueError: mutable default <class 'flytekit.configuration.S3Config'> for field s3 is not allowed: use default_factory

Hi team, I found that executing tasks with out no parameter renders an error in FlyteAdmin:

failed to create workflow in propeller json: error calling MarshalJSON for type *v1alpha1.Inputs: Marshal called with nil

Is this expected?

Hi guys. I am trying to run a task in sandbox that reads and writes data to an external sql db running in another container. How can I make this work? sorry if this is a silly question but I am pretty new with flyte. Thank you!

How do I implement dynamic epochs (e.g. early stopping) with flyte? Option A: define

train

and

eval

tasks and run them as long es condition is met in

@dynamic

@task
def train():
    ...

@task
def eval():
    ...

@dynamic
def job():
    running = True
    
    while running:
        train_result = train()

        score = eval()

        if score > 0.5:
            running = False

Option B: One "giant" workflow without

train

and

eval

tasks:

def train():
    ...

def eval():
    ...

@workflow
def job():
    running = True
    
    while running:
        train_result = train()

        score = eval()

        if score > 0.5:
            running = False

Are there advantages of one vs. the other? Would it be problematic that in Option A the DAG might grow very big?

@Samhita Alla in the video you linked, they briefly say you could implement "periodic triggers" using

sleep

(attached screenshot). But there is no example given, unfortunately. Therefore, I'm asking here again: How would you use

sleep

to trigger a task periodically inside a `@dynamic`workflow and how would the DAG look like in that case?

Anyone seen this error when trying to launch a workflow from the UI, it appeared after clicking "Launch Workflow" after redefining the workflow's input arguments to include a FlyteFile:

We are setting up flyte in GCP, is ingress and FQDN a mandatory requirement for a production ready setup? can we can run the jobs without having a FQDN?

hi Everyone… I have installed flyte admin in AWS, I can access without problem with a custom url.. using that url in my config to register a package I get the following error: transport: Error while dialing dial tcp: address [url]: missing port in address

really needs to have a port open?

currently only 80 and 443 are open

RFC for tags and execution description support in Flyte - https://github.com/flyteorg/flyte/pull/3320. Please TAL and comment

Is it possible to cleanup project instead of archiving them? As in remove from DB and remove k8s namespaces. It looks like scheduler tries to schedule even archived projects and spits out errors

Got error while scheduling rpc error: code = InvalidArgument desc = project [project-x] is not active

I’m seeing a strange error whilst executing a fairly nested workflow:

Workflow[<worfklow-id>] failed. RuntimeExecutionError: max number of system retry attempts [51/50] exhausted. Last known status message: Failed to check Catalog for previous results: DataCatalog failed to get dataset for ID resource_type:TASK name:"<task-id>" version:"<task-version>" : rpc error: code = InvalidArgument desc = missing project

This seems to come from here, but I’m not quite sure why

project

would be empty here

Hey guyz, I did a setup of flyte on GCP. I'm able to access the UI but on running pyflyte run --remote sample_workflow.py main --x 10 --y 20 I'm getting this error,

grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
        status = StatusCode.UNAVAILABLE
        details = "failed to connect to all addresses; last error: UNKNOWN: ipv4:172.25.0.46:443: Ssl handshake failed: SSL_ERROR_SSL: error:1000007d:SSL routines:OPENSSL_internal:CERTIFICATE_VERIFY_FAILED"
        debug_error_string = "UNKNOWN:failed to connect to all addresses; last error: UNKNOWN: ipv4:172.25.0.46:443: Ssl handshake failed: SSL_ERROR_SSL: error:1000007d:SSL routines:OPENSSL_internal:CERTIFICATE_VERIFY_FAILED {grpc_status:14, created_time:"2023-02-07T18:47:06.951145+05:30"}"

Any body has an idea what is the problem here? P.s I have TLS disabled on ingress.

Hi, I've been running into "Death by etcd timeouts" situations on AKS. Symptoms mostly propeller unable to ListAndWatch fly crd objects thus -> halt and catch fire. Tried to remedy the situation by cleaning up what I think flyte's garbage collection is supposed to clean up and didnt (fly crd objects and task-pods), this worked once, but eventually the etcd behind AKS isn't responding any more. the number of flyteworkflow crd objects in etcd is in the 10k / 100k range after a couple days eventually. the version of flytadmin we're using is based on v1.1.39 and includes changes from this pr: https://github.com/flyteorg/flyteadmin/pull/504 however garbage collection works fine on our other AKS-based environments most of the time but it definitely looks unreliable sometimes, as in very irregularly cleans up. is this something we can configure/tune maybe? to really aggressively do garbage collection?

Hello! Is there a way to integrate Flyte with DataHub? Has anyone here did this?

Another question: How can I use Data Persistence Layer in GCS? I had looked on the documentation, but found nothing... Is there an example of some sort?

I can set annotations when I execute a workflow in the flyte UI, but I don't see how to do it via flytectl create execution. It looks like the command doesn't accept annotations in its config struct. Is that correct, or am I missing something?

Hello, I have a simple workflow with dataclass_json as the input:

from flytekit import workflow, LaunchPlan, task
from dataclasses import dataclass
from dataclasses_json import dataclass_json

@dataclass_json
@dataclass
class MyDataClass(object):
    def __init__(self, a: int, b: str):
        self.a = a
        self.b = b

@task
def example_node(input: MyDataClass):
    print(f'{input.a}')
    

@workflow
def my_dataclass_workflow(input: MyDataClass):
    return example_node(input=input)

Should the UI be editable to provide input as json?

Hi everyone.. any familiar with this error running a workflow? [1/1] currentAttempt done. Last Error: USER::Pod failed. No message received from kubernetes. [a8crkvcv4mzm69kk5hp5-n0-0] terminated with exit code (1). Reason [Error]. Message: exec /opt/env/bin/pyflyte-execute: exec format error .

added an integration to this channel: DevRel

Hi community, are there plans on supporting more complicated logic for conditions. As noted on the Flyte documentation:

Conditions are limited to certain binary and logical operators and can only be performed on primitive values.

Hi Flyte community, What is the default username/password of the local minio server created by

flytectl demo start

? I would like to login to it to check out the metadata stored there during task/workflow registration.

Folks comment on https://www.linkedin.com/posts/josh-tobin-4b3b10a9_what-are-peoples-favorite-tools-for-training-activity-7028506088647888896-aDb8?utm_source=share&amp;utm_medium=member_ios

@Sujith Samuel / @Ankit Goyal / @honnix / @Klaus Azesberger / @Bernhard Stadlbauer / @Rahul Mehta if some of you can help me test a new authentication engine in flytekit. This would really help us merge the code faster. I think some of you have unique usecases - https://github.com/flyteorg/flytekit/pull/1458