hi everybody, I am currently using

prefect cloud

, and my flow will invoke

aws lambda

and return me with the response. However, after registering my flow, whenever i try to run it, it throws

Unexpected error: TypeError("cannot serialize '_io.BufferedReader' object")

Is there a workaround on this?

@task(name="invoke_lambda")
def invoke_lambda(function_name, table_path, etl_target_date):
    lambda_client = boto3.client("lambda")
    response = lambda_client.invoke(
        FunctionName=function_name,
        Payload=json.dumps({"table_path": table_path, "etl_target_date": etl_target_date}),
    )
    return response


with Flow(
    "test-flow",
    executor=LocalExecutor(),
    run_config=LocalRun(),
    storage=S3(
        bucket="random-bucket",
    ),
) as flow:
    x = invoke_lambda("test", "a/b/c", "2021/04/02")


flow.register(project_name="xxx", labels=["dev"])

Hi Jeremy - this suggests that boto3’s response object is not serializable (very common in boto3); you need to figure out how to convert the response to a basic Python object - perhaps there is a to_json() method or something like that?

Prefect also has a handy

LambdaInvoke

task built in https://github.com/PrefectHQ/prefect/blob/master/src/prefect/tasks/aws/lambda_function.py#L219

I have also tried the lambdaInvoke task by prefect but it also has the same issue

Can you do something like this? (this worked for me) Just read the response content. I also had the same error as you without reading it.

import prefect
import json
import boto3
from prefect import task, Flow

@task(name="invoke_lambda")
def invoke_lambda(function_name):
    lambda_client = boto3.client("lambda")
    response = lambda_client.invoke(
        FunctionName=function_name,
        Payload=json.dumps({}),
    )
    return response['Payload'].read()

with Flow("lambda-test") as flow:
    x = invoke_lambda("TestingLambdaInvoke")

flow.register(project_name="example-project")

hi, i encountered something similar, my code is as follows:

import json

import boto3
from prefect import Flow, task
from prefect.executors import LocalDaskExecutor
from prefect.storage import S3


@task
def extract():
    return 8


@task
def transform(x):
    session = boto3.Session()
    lambda_client = session.client("lambda")

    response = lambda_client.invoke(
        FunctionName="my-function", Payload=json.dumps({"number": x})
    )

    return response


with Flow(
    name="compute-flow",
    executor=LocalDaskExecutor(),
    # schedule=Schedule(clocks=[CronClock("*/5 * * * *")]),
    storage=S3(
        bucket="my-bucket",
    ),
) as flow:
    e = extract()
    t = transform(e)

I believe this is more related to task results needing to be serializable for it to be distributed to Dask workers: https://docs.prefect.io/core/concepts/common-pitfalls.html#serialization

thanks kevin, let me give it a try right away

thanks for telling me! i was making a test myself

now i understand chris's statement

yes that's exactly right

a follow up question,

from prefect.utilities.debug import is_serializable


is_serializable(my_flow) # returns True / False

^ i can use this to check whether the entire flow is serializable?

Yes

alright! thanks for pointing me to the right direction.

No problem. Happy to help 🙂

@Kevin Kho thank you 😄. will give it a try!