Hi Team, I was going through the external resources implementation in Flink. Our usecase is we have to to model inference over the data that we receive, which requires GPU. As per the doc we have tried implementing the same, by setting up the values for nvidia. What we have observed is that - when we setup the amount = 1, each task manager that we spawn will start taking the entire GPU. This inturn is causing a critical resource scaling issue - say if we want 10TM then we need 10GPU's which is pretty costly. Is there a way by which we can solve this case, any kind of resource sharing mechanisms, or do we need to go ahead with a model server architecture. Note - we are using Nvidia T4 GPU nodes where CPU=4 , memory=28GB #C065944F9M2 #C03G7LJTS2G

Define task slots per task manager is equal to = 4 ( number of CPU's). If you are using not much CPU bound work loads you can define task slots =8. Then Its all boils down to paralleism of your job. If you defined paralleism of job is 8, then you can define replicas of task manager:1. If you define16, define replicas of task manager 2, total 2 task managers. => replicas of task anager= Total Taskslots/Job paralleism. This will help you to control task managers

@Vijaya Bhaskar V the models that we are running is both CPU and GPU bounded. From the Nvidia T4 GPU nodes where CPU=4 , memory=28GB -- the allocatable CPU is ~2.75. What we are trying to figure out is there any efficient way of utilizing the Flink external resources for GPU usage. Based on what we have tried out, our understanding is that the GPU resources will be allocated per TM (we are keeping the replicas as 1 and slots as 1 due to the CPU requirement), which seems a bottleneck when we want to scale up.

flink run-application \
        -t kubernetes-application \
        -Dkubernetes.deployment.mode=application \
        -p $PARALLELISM \
        -Dkubernetes.namespace=$FLINK_NAMESPACE \
        -Dkubernetes.cluster-id=$APPLICATION \
        -Dkubernetes.container.image=$FLINK_IMAGE \
        -Dkubernetes.taskmanager.node-selector=$NODE_LABEL \
        -Dkubernetes.taskmanager.tolerations="key:platform,operator:Equal,value:gpu,effect:NoSchedule" \
        -Dkubernetes.pod-template-file.jobmanager=./flink-streaming/k8s/jobmanager.yaml \
        -Dkubernetes.pod-template-file.taskmanager=./flink-streaming/k8s/taskmanager.yaml \
        -Dkubernetes.jobmanager.cpu=1 \
        -Dkubernetes.taskmanager.registration.timeout=10min \
        -Dkubernetes.taskmanager.cpu=2.75 \
        -Dkubernetes.taskmanager.replicas=1 \
        -Dtaskmanager.numberOfTaskSlots=1 \
        -Dparallelism.default=$PARALLELISM \
        -Djobmanager.memory.process.size=4g \
        -Dtaskmanager.memory.process.size=10g \
        -Dtaskmanager.memory.task.off-heap.size=512m \
        -Dtaskmanager.memory.framework.off-heap.size=128m \
        -Dtaskmanager.memory.managed.size=768m \
        -Dtaskmanager.memory.network.min=256m \
        -Dtaskmanager.memory.network.max=512m \
        -Dtaskmanager.jvm-options="-XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:G1HeapRegionSize=16m -XX:+PrintGCDetails" \
        -Dheartbeat.timeout=300000 \
        -Dheartbeat.interval=10000 \
        -Dakka.ask.timeout=120s \
        -Dpython.fn.execution.bundle.size=1 \
        -Dpython.fn.execution.bundle.time=1000 \
        -Dkubernetes.cluster.config.map.name=flink-config-pyflink-job \
        -Dstate.backend=rocksdb \
        -Dstate.backend.rocksdb.block.cache-size=256mb \
        -Dstate.backend.rocksdb.write-buffer-size=64mb \
        -Dstate.backend.rocksdb.max-write-buffer-number=2 \
        -Dstate.backend.incremental=true \
        -Dstate.backend.rocksdb.ttl.compaction.filter.enabled=true \
        -Dstate.backend.rocksdb.compaction.level.use-dynamic-size=true \
        -Dexternal-resources=gpu \
        -Dexternal-resource.gpu.kubernetes.config-key=$GPU_KEY \
        -Dstate.checkpoints.dir=wasbs:azure_path \
        -Dfs.azure.account.key.${AZURE_STORAGE_ACCOUNT}.<http://blob.core.windows.net|blob.core.windows.net>=${AZURE_STORAGE_KEY} \
        -Dmetrics.reporters=prom \
        -Dmetrics.reporter.prom.factory.class=org.apache.flink.metrics.prometheus.PrometheusReporterFactory \
        -Dmetrics.reporter.prom.port=9249 \
        -py /opt/flink/flink_job/ppe_adherence_deep_tracker/main.py
}

See the moment we keep task slots =1 then it considers only 1 task manager per machine. Thats the behavior of flink. You might have defined paralleism: 10 for the job thats why 10 task managers created, each using a machine

@Vijaya Bhaskar V i understood the point, yes in our case since the CPU required is 2 cores per slot, this is because of the model that we are running. Is there any possibility by which we can efficiently use the GPU resources. Like we will create 2 TM one with CPU and other with GPU and in our flink code whenever needed the stages / tasks can be shifted to the required TM

See i got your issue: For GPU sharing and running application must know how to do it. AFAIK, For flink currently there is no application level handling how to use GPU efficiently, like using CUDA libraries.. Where as for CPU application doesn't need to since its handled by programming language level