Hi everyone, TLDR; I have some performance issue and I want to better understand how to properly set-up parallelism within Flink. Use case: I want to reflect every changes happening in my MSSQL views into Kafka topics . Therefore, I use Flink to re-create the MSSQL views based on different MSSQL tables. Overall my pipeline simply looks like this:

MSSQL --> Flink --> Kafka

Flink is running in Kubernetes, and I'm using the [flink-sql-runner](https://github.com/apache/flink-kubernetes-operator/tree/main/examples/flink-sql-runner-example) in order to be able to simply provide SQL scripts to the Flink Kubernetes Operator. Issue: It's very slow. It takes ~20minutes to produce ~10K messages in Kafka. The input data aren't that big (3 tables: 10K rows, 300 rows, 25millions rows), and the computation is quite easy: the view consists of a simple combination of an INNER JOIN and an LEFT OUTER JOIN. I notice that my job goes very busy and have much backpressure (100% sometime). Open question: I'm trying to make sense of the parameters

taskmanager.numberOfTaskSlots

,

table.exec.resource.default-parallelism

and

parallelism

in my K8s deployment (see below). What is the difference between

table.exec.resource.default-parallelism

and

parallelism

(under the "job" section) ? I suppose the

cpu

should be set accordingly to the parallelism/numberOfTaskSlots; what is the best practice there between those parameters?

apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkDeployment
metadata:
  name: foobar
spec:
  image: flink-sql-runner:latest
  flinkVersion: v1_17
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
    execution.checkpointing.interval: 10s
    state.backend: rocksdb
    state.backend.incremental: "true"
    state.savepoints.dir: "<s3://flink/savepoints/sql-example/>"
    state.checkpoints.dir: "<s3://flink/checkpoints/sql-example/>"
    high-availability.storageDir: "<s3://flink/recovery/>"
    high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
    containerized.master.env.ENABLE_BUILT_IN_PLUGINS: "flink-s3-fs-hadoop-1.17.1.jar"
    containerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS: "flink-s3-fs-hadoop-1.17.1.jar"
    s3.access-key: "0"
    s3.secret-key: "0"
    s3.endpoint: "<http://host.docker.internal:4566>"
    s3.path.style.access: "true"
    table.exec.resource.default-parallelism: "4"
  serviceAccount: flink
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
    podTemplate:
      spec:
        containers:
          - name: flink-main-container
            volumeMounts:
              - name: sql-scripts
                mountPath: /opt/sql-scripts
        volumes:
          - name: sql-scripts
            configMap:
              name: sql-scripts
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1
  job:
    jarURI: local:///opt/flink/usrlib/sql-runner.jar
    args: ["/opt/sql-scripts/my-statement.sql"]
    parallelism: 1
    upgradeMode: last-state
    state: running
    savepointTriggerNonce: 0

@Tiphanie Dousset, You can check flameGraph while running job where the bottleneck is