Hi team.. I need some help. I have a flyte task that needs to read a parquet file stored in s3 (about 2 GB in size, 100k rows) and return 1000 records from it… I’m using awswrangler for reading the file into a pandas dataframe and then return the first 1000 records from it (list of dicts). Each task node has upto 8 GB in memory… but the pod keeps getting OOM killed with no useful info in the stack trace. Not sure what’s going on…

is 8GB task resource request or limit?

sorry.. not 8.. 18 Here’s a snippet of our helm chart

# -- Task default resources configuration
  task_resource_defaults:
    task_resources:
      defaults:
        cpu: 100m
        memory: 200Mi
        storage: 200Mi
        ephemeral-storage: 1Gi
        gpu: 0
      limits:
        cpu: 9
        memory: 18Gi
        storage: 20Gi
        ephemeral-storage: 28Gi
        gpu: 0

how about the pod spec? could you show me the output of kubectl get pods <name> -n flytesnack-development

Sent you the screen shot… Meanwhile here are the sanitized logs from a pod that got OOMKilled

{"asctime": "2022-12-28 05:20:19,825", "name": "flytekit", "levelname": "INFO", "message": "Setting protocol to file"}
{"asctime": "2022-12-28 05:20:25,924", "name": "flytekit", "levelname": "INFO", "message": "We won't register PyTorchCheckpointTransformer, PyTorchTensorTransformer, and PyTorchModuleTransformer because torch is not installed."}
{"asctime": "2022-12-28 05:20:26,124", "name": "flytekit", "levelname": "INFO", "message": "Setting protocol to file"}
{"asctime": "2022-12-28 05:20:26,124", "name": "flytekit", "levelname": "INFO", "message": "Setting protocol to file"}
{"asctime": "2022-12-28 05:20:26,125", "name": "flytekit", "levelname": "INFO", "message": "Setting protocol to file"}
{"asctime": "2022-12-28 05:20:26,125", "name": "flytekit", "levelname": "INFO", "message": "Setting protocol to file"}

{"asctime": "2022-12-28 05:20:27,733", "name": "flytekit.entrypoint", "levelname": "INFO", "message": "Welcome to Flyte! Version: 1.2.1"}
{"asctime": "2022-12-28 05:20:27,734", "name": "flytekit.entrypoint", "levelname": "INFO", "message": "Using user directory /tmp/flyte-i8059ewt/sandbox/local_flytekit/9f57ef630f6df792206a67f0b4e593c5"}
{"asctime": "2022-12-28 05:20:32,830", "name": "flytekit", "levelname": "INFO", "message": "Entering timed context: Copying (<s3://my-flyte/flyte/metadata/propeller/project-development-agbd42n8ph94v8rwcbgv/n0/data/inputs.pb> -> /tmp/flyte-i8059ewt/sandbox/local_flytekit/inputs.pb)"}
{"asctime": "2022-12-28 05:20:41,531", "name": "flytekit", "levelname": "INFO", "message": "Output of command '['aws', 's3', 'cp', '<s3://my-flyte/flyte/metadata/propeller/project-development-agbd42n8ph94v8rwcbgv/n0/data/inputs.pb>', '/tmp/flyte-i8059ewt/sandbox/local_flytekit/inputs.pb']':\nb'Completed 130 Bytes/130 Bytes (697 Bytes/s) with 1 file(s) remaining\\rdownload: <s3://my-flyte/flyte/metadata/propeller/project-development-agbd42n8ph94v8rwcbgv/n0/data/inputs.pb> to ../tmp/flyte-i8059ewt/sandbox/local_flytekit/inputs.pb\\n'\n"}
{"asctime": "2022-12-28 05:20:41,532", "name": "flytekit", "levelname": "INFO", "message": "Exiting timed context: Copying (<s3://my-flyte/flyte/metadata/propeller/project-development-agbd42n8ph94v8rwcbgv/n0/data/inputs.pb> -> /tmp/flyte-i8059ewt/sandbox/local_flytekit/inputs.pb) [Wall Time: 8.701755479996791s, Process Time: 0.027927802s]"}
{"asctime": "2022-12-28 05:20:41,534", "name": "flytekit", "levelname": "INFO", "message": "Invoking flyte.workflows.my_workflow.read_input_records with inputs: {'num_requests': 1000, 'input_path': '<s3://my_bucket/some_file_name.parquet>'}"}

sorry, I said something wrong. should be kubectl describe …

I’m seeing this in the describe pod output

Limits:
      cpu:     100m
      memory:  200Mi
    Requests:
      cpu:     100m
      memory:  200Mi

could you check the propeller config map? just want to make sure the resource is 18GB in that configmap.

python task

no, kubectl edit cm <propeller config map>

ah.. hang on

sorry again, flyteadmin config map

kubectl edit cm flyte-admin-base-config

task_resource_defaults.yaml: |
    task_resources:
      defaults:
        cpu: 100m
        ephemeral-storage: 1Gi
        gpu: 0
        memory: 200Mi
        storage: 200Mi
      limits:
        cpu: 9
        ephemeral-storage: 28Gi
        gpu: 0
        memory: 18Gi
        storage: 20Gi

one sec

no… didn’t realize I needed to do that

no, you don’t. just to confirm you didn’t change that

is it the task_config parameter?

yes. https://github.com/flyteorg/flytekit/blob/59443b8f4a6a8a1ba6015b879b7324db3ab53d36/plugins/flytekit-k8s-pod/tests/test_pod.py#L254

thanks.. so do I set both?

I think you only need to set limit. if it doesn’t work, try to set both

No crash so far 🤞🏼

@little-cricket-84530 can you help Document what you had to do

Sure.. but not sure why on earth that task took 4.5 hours??? All it does is read a 2 GB parquet file into a pandas data frame and return the first 1000 records. From the logs there appears to be a lot of reading and writing to s3 in tiny chunks (600 bytes here, 7 KB there and so on) Example

Exiting timed context: Writing (/tmp/flyte2qtcr_i0/local_flytekit/f4484561b67506b8c8bfe18c1b4a315d -> <s3://my-flyte/2x/axd9n8qhbwlgszfb9slf-n0-0/939989a9a40c75f65c631bdf11070354/f4484561b67506b8c8bfe18c1b4a315d>) [Wall Time: 6.298165863845497s, Process Time: 0.029386856000002126s]
1614	Output of command '['aws', 's3', 'cp', '--acl', 'bucket-owner-full-control', '/tmp/flyte2qtcr_i0/local_flytekit/f4484561b67506b8c8bfe18c1b4a315d', '<s3://my-flyte/2x/axd9n8qhbwlgszfb9slf-n0-0/939989a9a40c75f65c631bdf11070354/f4484561b67506b8c8bfe18c1b4a315d']'>:
b'Completed 8.9 KiB/8.9 KiB (47.7 KiB/s) with 1 file(s) remaining\rupload: ../tmp/flyte2qtcr_i0/local_flytekit/f4484561b67506b8c8bfe18c1b4a315d to <s3://tesla-flyte/2x/axd9n8qhbwlgszfb9slf-n0-0/939989a9a40c75f65c631bdf11070354/f4484561b67506b8c8bfe18c1b4a315d>\n'
1615	Entering timed context: Writing (/tmp/flyte2qtcr_i0/local_flytekit/8016c26332f3085b27a3d08692b5df94 -> <s3://my-flyte/2x/axd9n8qhbwlgszfb9slf-n0-0/ebb807108d3cd57bbfff58b978a8fa22/8016c26332f3085b27a3d08692b5df94>)

Might be Flyte deck issue

let me try that.. I saw that on another thread but wasn’t sure if it applied to my use case

btw, is fsspec package installed in your image? There are some issues in fsspec we just found that. it will download the files serially

yes.. I do have fsspec

sorry, could you remove that? and then try to run the workflow again. I’m going to fix the bug in fsspec plugin

I’ll remove it for now in my branch.. since this specific workflow doesn’t need it..

thanks

@glamorous-carpet-83516 it looks like the bottleneck is writing to s3/flyte’s ephemeral storage when the task returns.. at this point it’s just 1000 rows

hmm this is really slow

15 min and counting….

does not make sense

I’m looking into it

I’ll be happy to share the code of the task (nothing proprietary in there).. if you want to dig in

Yes, that help

import awswrangler as wr
import pandas as pd
import boto3
from time import perf_counter

@task(
    #cache=True,
    #cache_version="1.0",
    limits=Resources(mem="10Gi"),
    disable_deck=True,
    secret_requests=[
        Secret(group=SECRET_GROUP, key="aws_secret_access_key"),
        Secret(group=SECRET_GROUP, key="aws_access_key_id"),
    ],
)
def read_input_records(input_path: str, num_requests: int) -> List[Dict[str, Any]]:
    aws_access_key_id = flytekit.current_context().secrets.get(
        SECRET_GROUP, "aws_access_key_id"
    )
    aws_secret_access_key = flytekit.current_context().secrets.get(
        SECRET_GROUP, "aws_secret_access_key"
    )
    session = boto3.Session(
        aws_access_key_id=aws_access_key_id,
        aws_secret_access_key=aws_secret_access_key,
        region_name=AWS_REGION,
    )
    start = perf_counter()
    df = wr.s3.read_parquet(path=input_path, boto3_session=session)
    stop = perf_counter()
    flytekit.current_context().<http://logging.info|logging.info>(
        f"Read {len(df)} records in {stop - start} s"
    )

    if num_requests != -1 and num_requests < len(df):
        df = df[:num_requests]
    stop_2 = perf_counter()
    flytekit.current_context().<http://logging.info|logging.info>(
        f"Truncated to {len(df)} records in {stop_2 - stop} s"
    )

    records = df.to_dict("records")

    stop_3 = perf_counter()
    flytekit.current_context().<http://logging.info|logging.info>(
        f"Converted to records in {stop_3 - stop_2} s"
    )

    return records

Please adjust aws credentials etc as per your setup

Has the task been completed ? IIRC, if you use a large list (10000+) as output, flyte will take a long time (15 mins+) to construct a large protobuf list, and save it. However, saving size 1000 of a list should be really fast What is the number of columns in the dataframe? The record should be 1000 (list size)*N (number of column), right? To download the file in the pod, awswrangler should spend an equal amount of time as local, so the issue might only happen when we save the record or construct a large protobuf list

thanks a lot.. ok.. let me adjust the type

Ohh we need to somehow show this

I think there was some message about using Pickle for typing.Any… but would be nice to throw this into the docs somewhere as a real warning

As for the task_resource limits that was brought up earlier in this thread

task_resource_defaults.yaml: |
    task_resources:
      defaults:
        cpu: 100m
        ephemeral-storage: 1Gi
        gpu: 0
        memory: 200Mi
        storage: 200Mi
      limits:
        cpu: 9
        ephemeral-storage: 28Gi
        gpu: 0
        memory: 18Gi
        storage: 20Gi

The

task_resources.defaults

refers to the defaults limits that apply on any POD. The

task.resources.limits

refers to the upper limit that can be set from the user's repo. If no request/limit is set on a task, the task_resources.defaults are applied as both. Users cannot set higher limits for their tasks than the

task_resources.limits

.

@swift-animal-75798 Thanks for the explain.