AIP-7487 join memory fix

[talebzeghmi]

• reduce s3op NUM_WORKERS_DEFAULT on a join (loading data from parent steps/splits) from 64 to 5
• note: This only increased the time in order of seconds, albeit this is with lightweight small s3 self data.
• This can be overridden in a join step with the following env decorator

@environment(  # pylint: disable=E1102
  vars={"METAFLOW_S3OP_NUM_WORKERS_DEFAULT": "3"}
)
@step

Test run

import time
from metaflow import FlowSpec, step, resources, exit_handler, Parameter, environment


class HelloFlow(FlowSpec):
    splits = Parameter("splits", default=10)

    @step
    def start(self):
        print("HelloFlow is starting.")
        self.x = "foo"
        self.y = "foo"
        self.z = "foo"
        self.a = "b"
        self.my_list = list(range(self.splits))
        self.next(self.compute, foreach="my_list")

    @step
    def compute(self):
        self.c = self.input
        time.sleep(2)
        self.next(self.join)

    @environment(  # pylint: disable=E1102
        vars={"METAFLOW_S3OP_NUM_WORKERS": "10"}
    )
    @step
    def join(self, inputs):
        self.merge_artifacts(inputs, exclude=["c"])
        print("done")
        self.next(self.end)

    @step
    def end(self):
        print("HelloFlow is all done.")


if __name__ == "__main__":
    HelloFlow()

Run 1 w/ 100 splits METAFLOW_S3OP_NUM_WORKERS=10

• python metaflow/tutorials/00-helloworld/helloworld.py aip run --splits=100 --max-parallelism 30
• setting METAFLOW_S3OP_NUM_WORKERS=10
• max memory usage of 412MiB
  
• https://ai-platform.datadoghq.com/dashboard/kum-8g3-6gj/demo---ai-platform-workflow-dashboard-cloned?refresh_mode=paused&tpl_var_kube_namespace%5B0%5D=aip-playground-dev&tpl_var_pod_name%5B0%5D=helloflow-jnz9m-3018403840&tpl_var_rund_id%5B0%5D=argo-helloflow-jnz9m&from_ts=1701971570539&to_ts=1701971870539&live=false
• https://argo-server.int.stage-k8s.zg-aip.net/argo-ui/workflows/aip-playground-dev/helloflow-jnz9m

Run 2 w/ 100 splits METAFLOW_S3OP_NUM_WORKERS=5

With

• python metaflow/tutorials/00-helloworld/helloworld.py aip run --splits=100 --max-parallelism 30
• max memory usage of 269MiB
  
• https://ai-platform.datadoghq.com/dashboard/kum-8g3-6gj/demo---ai-platform-workflow-dashboard-cloned?refresh_mode=paused&tpl_var_kube_namespace%5B0%5D=aip-playground-dev&tpl_var_pod_name%5B0%5D=helloflow-prhxk-4170305833&tpl_var_rund_id%5B0%5D=argo-helloflow-prhxk&from_ts=1701973343834&to_ts=1701973643834&live=false
• https://argo-server.int.stage-k8s.zg-aip.net/argo-ui/workflows/aip-playground-dev/helloflow-prhxk