Hi Dongwon, If you know the data in advance, you can always use the Yarn options in [1] (e.g. the "yarn.ship-directories") to ship the directories with the data you want only once to each Yarn container (i.e. TM) and then write a udf which reads them in the open() method. This will allow the data to be shipped only once per TM but then each of the tasks will have its own copy in memory of course. By default the visibility of the files that you ship is set to APPLICATION [2], if I am not mistaken so if more than one TMs go to the same node, then you will have even less copies shipped.
Does this help with your usecase? Cheers, Kostas [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/config.html [2] https://hadoop.apache.org/docs/stable/api/org/apache/hadoop/yarn/api/records/LocalResourceVisibility.html On Sun, Sep 20, 2020 at 6:05 PM Dongwon Kim <eastcirc...@gmail.com> wrote: > > Hi, > > I'm using Flink broadcast state similar to what Fabian explained in [1]. One > difference might be the size of the broadcasted data; the size is around > 150MB. > > I've launched 32 TMs by setting > - taskmanager.numberOfTaskSlots : 6 > - parallelism of the non-broadcast side : 192 > > Here's some questions: > 1) AFAIK, the broadcasted data (150MB) is sent to all 192 tasks. Is it right? > 2) Any recommended way to broadcast data only to 32 TMs so that 6 tasks in > each TM can read the broadcasted data? I'm considering implementing a static > class for the non-broadcast side to directly load data only once on each > TaskManager instead of the broadcast state (FYI, I'm using per-job clusters > on YARN, so each TM is only for a single job). However, I'd like to use Flink > native facilities if possible. > > The type of broadcasted data is Map<Long, Int> with around 600K entries, so > every time the data is broadcasted a lot of GC is inevitable on each TM due > to the (de)serialization cost. > > Any advice would be much appreciated. > > Best, > > Dongwon > > [1] https://flink.apache.org/2019/06/26/broadcast-state.html
