title: Please suggest hardware configuration for network-intensive Flink job (Async I/O)
body: <p>TLDR; I am running Flink Streaming job in mode=Batch on EMR. I have tried several EMR cluster configurations but neither of them works as required. Some do not work at all. Workflow is very network-intensive that cases main problems.</p>
  <p><strong>Question: What EMR cluster configuration (ec2 instance types) would you recommend for this use-case?</strong></p>
  <p>--</p>
  <p>The job has following stages:</p>
  <ol>
  <li>Read from MySQL</li>
  <li>KeyBy user_id</li>
  <li>Reduce by user_id</li>
  <li>Async I/O enriching from Redis</li>
  <li>Async I/O enriching from other Redis</li>
  <li>Async I/O enriching from REST #1</li>
  <li>Async I/O enriching from REST #2</li>
  <li>Async I/O enriching from REST #2</li>
  <li>Write to Elasticsearch</li>
  </ol>
  <p>Other info:</p>
  <p>Flink version: 1.13.1
  EMR version: 6.4.0
  Java version: JDK version Corretto-8.302.08.1 (provided by EMR)
  Input data size: ~800 GB
  Output data size: ~300 GB</p>
  <ul>
  <li>&quot;taskmanager.network.sort-shuffle.min-parallelism&quot;: 1</li>
  <li>&quot;taskmanager.memory.framework.off-heap.batch-shuffle.size&quot;: 256m</li>
  <li>&quot;taskmanager.network.sort-shuffle.min-buffers&quot;: 2048</li>
  <li>&quot;taskmanager.network.blocking-shuffle.compression.enabled&quot;: true</li>
  <li>&quot;taskmanager.memory.framework.off-heap.size&quot;: 512m</li>
  <li>&quot;taskmanager.memory.network.max&quot;: 2g</li>
  </ul>
  <p>Configurations we tried:</p>
  <p><strong>#1</strong></p>
  <p>master: r6g.xlarge</p>
  <p>core: r6g.xlarge (per/hour: $0.2; CPU: 4; RAM: 32 GiB; Disk: EBS 128 GB, network: 1.25 Gigabit baseline with burst up to 10 Gigabit)</p>
  <p>min_scale: 2</p>
  <p>max_scale: 25</p>
  <ul>
  <li>expected: finishes within 24 hours</li>
  <li>actual: works with <a href="https://flink.apache.org/2021/10/26/sort-shuffle-part1.html" rel="nofollow noreferrer">sort-based shuffling</a> enabled but very slowly (~36h), as this type of instance has a baseline &amp; burst performance, when burst credits are exhausted degrades to the baseline of 1GBps, that slows down I/O. With hash-based shuffling fails on KeyBy -&gt; Reduce with &quot;Connection reset by peer&quot;, Task Manager fails -&gt; Job fails -&gt; Job manager is not able to restart.</li>
  </ul>
  <p><strong>#2</strong></p>
  <p>master: m5.xlarge</p>
  <p>core: r6g.12xlarge (per/hour: $2.4; CPU: 48; RAM: 384 GiB; Disk: EBS 1.5 TB, network: 20 Gigabit)</p>
  <p>min_scale: 1</p>
  <p>max_scale: 4</p>
  <ul>
  <li>expected: finishes within 24 hours, as there is much higher network badwith</li>
  <li>actual: does not work. With sort-based shuffling fails on the writing phase with exception &quot;Failed to transfer file from TaskExecutor&quot;. With hash-based shuffling fails on the same stage with &quot;Connection reset by peer&quot;.</li>
  </ul>