class model_wrapper:
    @task
    def fit_classifier(self, train: pd.DataFrame,
                       val: pd.DataFrame) -> JoblibSerializedFile:

        # fetch the features and target columns from the train dataset
        x = train[train.columns[1:]]
        y = train[train.columns[0]]

        # fetch the features and target columns from the validation dataset
        eval_x = val[val.columns[1:]]
        eval_y = val[val.columns[0]]

        m = self.xgb_classifier()
        # fit the model to the train data
        m.fit(x, y, eval_set=[(eval_x, eval_y)])

        working_dir = flytekit.current_context().working_directory

        fname = os.path.join(working_dir, "model.joblib.dat")
        joblib.dump(m, fname)

        # return the serialized model
        return JoblibSerializedFile(path=fname)

...
raise _user_exceptions.FlyteAssertion("Input was not specified for: {} of type {}".format(k, var.type))
flytekit.exceptions.user.FlyteAssertion: Input was not specified for: self of type simple: NONE

flytectl config init

--config

~/.flyte/config.yaml

raise _InactiveRpcError(state)
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
	status = StatusCode.RESOURCE_EXHAUSTED
	details = "received initial metadata size exceeds limit"
	debug_error_string = "{"created":"@1667984783.531503108","description":"Error received from peer ipv4:10.162.12.35:443","file":"src/core/lib/surface/call.cc","file_line":953,"grpc_message":"received initial metadata size exceeds limit","grpc_status":8}"

FlyteFile

uri

Flytefile

kubernetes_service_account

LaunchPlan.get_or_create(
    name="my_lp",
    workflow=my_wf,
    auth_role=AuthRole(
        assumable_iam_role=None,
        kubernetes_service_account="my-flyte-service-account"
    ),
)

LaunchPlan.get_or_create(
    name="my_lp",
    workflow=my_wf,
    security_context=security.SecurityContext(
        run_as=security.Identity(
            iam_role="default",
            k8s_service_account="my-flyte-service-account",
        ),
    ),
)

@task
def t1(team: str, a: int) -> Tuple[str, int]:
    print(f'in t1() team: {team}')
    print(f'in t1() a: {a}')
    return team, a

@workflow
def wf(team: str = 'abc', a: int = 3) -> Tuple[str, int]:
    print(f'in wf() team: {team}')
    print(f'in wf() a: {a}')

    x, y = t1(team=f'{team}_{a}', a=a)
    return x, y

if __name__ == "__main__":
    print(f"result from wf(): {wf(team='abc', a=3)}")

in wf() team: abc
in wf() a: 3
in t1() team: abc_3
in t1() a: 3
result from wf(): ('abc_3', 3)

in wf() team: Promise(node:.team)
in wf() a: Promise(node:.a)
in wf() team: Resolved(team=scalar {
  primitive {
    string_value: "abc"
  }
}
)
in wf() a: Resolved(a=scalar {
  primitive {
    integer: 3
  }
}
)
in t1() team: Resolved(team=scalar {
  primitive {
    string_value: "abc"
  }
}
)_Resolved(a=scalar {
  primitive {
    integer: 3
  }
}
)
in t1() a: 3
result from wf(): DefaultNamedTupleOutput(o0='Resolved(team=scalar {\n  primitive {\n    string_value: "abc"\n  }\n}\n)_Resolved(a=scalar {\n  primitive {\n    integer: 3\n  }\n}\n)', o1=3)

Could not find a version that matches protobuf<4,>=3.5.0,>=3.6.1,>=4.21.6 (from flytekit==1.2.3

#15 13.29 There are incompatible versions in the resolved dependencies:
    #15 13.29   protobuf>=4.21.6 (from grpcio-status==1.50.0->flytekit==1.2.3->-r /root/requirements.in (line 1))
    #15 13.29   protobuf<4,>=3.6.1 (from flytekit==1.2.3->-r /root/requirements.in (line 1))
    #15 13.29   protobuf<4.0.0,>=3.5.0 (from flyteidl==1.1.22->flytekit==1.2.3->-r /root/requirements.in (line 1))
    #15 ERROR: executor failed running [/bin/sh -c pip-compile --output-file=/root/requirements.txt /root/requirements.in]: exit code: 2

>>

z >> x
TypeError: unsupported operand type(s) for >>: 'str' and 'DataFrame'

@task
def get_xgb_task_resource_request(df: pd.DataFrame) -> Resources:
...