< Marvin> upload combine excel is prefect task but it cash F Prefect Community #ask-marvin

<@ULVA73B9P> upload_combine_excel is prefect task...

Madushan Pathirana

10/21/2025, 10:00 AM

@Marvin upload_combine_excel is prefect task ,but it cash Finished in state Crashed('Execution was cancelled by the runtime environment.') this is the only log. @flow(name="EBAY",flow_run_name='{seller_id}-{data_cycle}', log_prints=True,task_runner=ThreadPoolTaskRunner(max_workers=8)) def run(seller_idstr,selling partner idstr,data_cycle:str): ebay_extraction = EbayExtraction(seller_id,selling_partner_id,data_cycle,extraction_period=7) dataframe_dict = {} refresh_token = ebay_extraction.get_marketplace() df_sellers = ebay_extraction.get_form() dataframe_dict['form'] = df_sellers.T full_name = df_sellers.loc[0,'full_name'] companyType = df_sellers.loc[0,'company_type'] run_pipeline = True if companyType in ['sole_proprietorship','individual_seller_unregistered']: run_pipeline = False if not run_pipeline: non_business_user_email_subject = f'Ebay Data Extraction Stoped - {seller_id}' non_business_user_email_body = f' Ebay data extraction stoped because user Company Type is {companyType} ' ebay_extraction.send_updates_as_email(non_business_user_email_subject,non_business_user_email_body ,recipient) else: form_email_subject = f'Ebay Data Extraction Started - {full_name}' ebay_extraction.send_dataframe_as_email(df_sellers.T,form_email_subject) user_df= ebay_extraction.get_user_data(refresh_token) seller_df = ebay_extraction.get_seller_data(refresh_token,user_df) stor_df , store_id = ebay_extraction.get_store_data(refresh_token) dataframe_dict['store'] = stor_df dataframe_dict['user'] = user_df dataframe_dict['seller'] = seller_df transactions_api = {"url" : "https://apiz.ebay.com/sell/finances/v1/transaction?", "upload_key" : 'transactions', "api_limit" : 200, "date_key": 'transactionDate'} payout_api = {"url" : "https://apiz.ebay.com/sell/finances/v1/payout?", "upload_key" : 'payouts', "api_limit" : 50, "date_key": 'payoutDate'} orders_api = {"url" : "https://api.ebay.com/sell/fulfillment/v1/order", "upload_key" : 'orders', "api_limit" : 200, "date_key": 'creationdate'} api_dict_list = [ transactions_api, payout_api, orders_api ] for api_dict in api_dict_list: upload_key = api_dict.get('upload_key') df = ebay_extraction.fetch_paginated_results.with_options( name="fetch_paginated_results", task_run_name=f"{upload_key}" ).submit(api_dict,refresh_token) dataframe_dict[upload_key] = df if upload_key == 'orders': # orders_df=df orders_lineitems_expanded_upload_key = 'order_with_lineItems' order_with_lineItems_expanded_df = ebay_extraction.orders_s3_upload.with_options( task_run_name=orders_lineitems_expanded_upload_key )(df,orders_lineitems_expanded_upload_key) dataframe_dict[orders_lineitems_expanded_upload_key] = order_with_lineItems_expanded_df seller_standards_profile_api = 'https://api.ebay.com/sell/analytics/v1/seller_standards_profile' seller_standards_profile_upload_key = 'seller_standards_profile' seller_standards_profile_api_metric='standardsProfiles' performance_df = ebay_extraction.get_perfomance( refresh_token,seller_standards_profile_api,seller_standards_profile_api_metric,seller_standards_profile_upload_key ) dataframe_dict[seller_standards_profile_upload_key] = performance_df ebay_extraction.upload_combine_excel.with_options( name="upload_combine_excel", task_run_name="upload_combine_excel" ).submit(dataframe_dict,store_id)

Marvin

10/21/2025, 10:00 AM

✅ thought for 302.7 seconds

Marvin

10/21/2025, 10:05 AM

Thanks for the snippet — that crash message is frustratingly terse. What “Finished in state Crashed('Execution was cancelled by the runtime environment.')” usually means - Prefect did not catch a normal Python exception in your task — the runtime (process/container) was terminated underneath it. Common causes: - The process/container was killed (e.g., out-of-memory, SIGKILL, ECS task stop) - The worker/runtime exited or was unavailable - Infrastructure limits/timeouts (for ECS, task memory/CPU/time) - In these cases, Prefect can only report a Crashed state with that generic message; there’s often no Python traceback. A few things I notice in your flow - You’re mixing

.submit(...)

and direct task calls — that’s okay in Prefect 3. If you pass a PrefectFuture into another task (direct call or

.submit

), Prefect will automatically resolve the future to its value. More details: Run work concurrently and Task runners - You’re passing a dict of large DataFrames/futures into

upload_combine_excel

. If that task is combining to Excel, it can be very memory hungry. An OOM kill will look exactly like this crash. Recommended next steps to isolate the root cause 1) Surface more detail in the flow by forcing resolution and logging - Capture the future for

upload_combine_excel

and call

.result()

to pull any failure into the flow logs (if it’s a normal failure, you’ll get a traceback; if it’s a crash, you’ll still just see crash but you’ll know it happens exactly there):

Copy code

u = ebay_extraction.upload_combine_excel.with_options(
    name="upload_combine_excel",
    task_run_name="upload_combine_excel"
).submit(dataframe_dict, store_id)

# Block to surface errors/crashes in flow logs
u.result(raise_on_failure=False)

- Consider resolving upstream futures before combining to isolate which data causes trouble:

Copy code

from prefect.futures import PrefectFuture

for k, v in list(dataframe_dict.items()):
    if isinstance(v, PrefectFuture):
        dataframe_dict[k] = v.result()  # will raise on failure and log a traceback

2) Check the runtime that actually executed the task - If you’re running on ECS: - Open CloudWatch logs for the ECS task/container around the crash time. Look for OOMKilled or SIGKILL. - Verify task memory/CPU settings. If you see OOM, increase memory limits or reduce memory usage in the task. - If you’re running locally on EC2: - Check the worker process logs (set

PREFECT_LOGGING_LEVEL=DEBUG

) and system logs (e.g., dmesg) for OOM kills. - Ensure the worker itself isn’t restarting or being terminated. 3) Add defensive logging inside

upload_combine_excel

- Log the number of sheets and the size of each DataFrame before writing. If you’re using pandas ExcelWriter, writing many large sheets can spike memory. - If possible, stream writes (write one sheet at a time and free memory) or consider more compact formats (CSV/Parquet) if Excel is not strictly required. 4) Be cautious with tasks defined as instance methods - If

upload_combine_excel

is an

@task

on a class instance (e.g.,

class EbayExtraction: @task def upload_combine_excel(self, ...)

), that can be fragile in some remote/infrastructure contexts. Prefer top-level

@task

functions or

@staticmethod

tasks to avoid serialization/import issues. Quick notes on Prefect 3 task semantics (to confirm your usage) -

Task.submit(...)

returns a

PrefectFuture

. Use

.result()

to get the Python value if you need it in-flow. Docs: Task runners - You can mix direct calls and

.submit

. Passing futures to other tasks is fine — Prefect resolves them automatically. Docs: Run work concurrently

Marvin

10/21/2025, 10:05 AM

- A Crashed state with “Execution was cancelled by the runtime environment.” indicates the runtime ended the work (not your code raising a normal exception). States docs: States A few questions to help me pinpoint this faster - Which work pool/worker type is running this flow? Local Process, Docker, ECS, etc.? - Do you see any ECS/CloudWatch or system logs that mention OOMKilled or task stopped? - Roughly how large are the DataFrames you’re writing to Excel (row counts, number of sheets)? - Can you share the

upload_combine_excel

task body (or at least how it writes the Excel)? If you share those details, I can suggest concrete changes (e.g., streaming Excel writing approach, memory tuning, or ECS task sizing) tailored to your setup.

5 Views

Open in Slack

Previous Next