FYI the Dataflow runner automatically sets the default number of shards
(I believe to be 2 * num_workers). Probably we should do something
similar for the Flink runner.
This needs to be done by the runner, as # of workers is a runner
concept; the SDK itself has no concept of workers.
On Thu, Oct 25, 2018 at 3:28 AM Jozef Vilcek <jozo.vil...@gmail.com
<mailto:jozo.vil...@gmail.com>> wrote:
If I do not specify shards for unbounded collection, I get
Caused by: java.lang.IllegalArgumentException: When applying
WriteFiles to an unbounded PCollection, must specify number of
output shards explicitly
at
org.apache.beam.repackaged.beam_sdks_java_core.com.google.common.base.Preconditions.checkArgument(Preconditions.java:191)
at
org.apache.beam.sdk.io.WriteFiles.expand(WriteFiles.java:289)
Around same lines in WriteFiles is also a check for windowed writes.
I believe FileIO enables it explicitly when windowing is present. In
filesystem written files are per window and shard.
On Thu, Oct 25, 2018 at 12:01 PM Maximilian Michels <m...@apache.org
<mailto:m...@apache.org>> wrote:
I agree it would be nice to keep the current distribution of
elements
instead of doing a shuffle based on an artificial shard key.
Have you tried `withWindowedWrites()`? Also, why do you say you
need to
specify the number of shards in streaming mode?
-Max
On 25.10.18 10:12, Jozef Vilcek wrote:
> Hm, yes, this makes sense now, but what can be done for my
case? I do
> not want to end up with too many files on disk.
>
> I think what I am looking for is to instruct IO that do not
do again
> random shard and reshuffle but just assume number of shards
equal to
> number of workers and shard ID is a worker ID.
> Is this doable in beam model?
>
> On Wed, Oct 24, 2018 at 4:07 PM Maximilian Michels
<m...@apache.org <mailto:m...@apache.org>
> <mailto:m...@apache.org <mailto:m...@apache.org>>> wrote:
>
> The FlinkRunner uses a hash function (MurmurHash) on each
key which
> places keys somewhere in the hash space. The hash space
(2^32) is split
> among the partitions (5 in your case). Given enough keys,
the chance
> increases they are equally spread.
>
> This should be similar to what the other Runners do.
>
> On 24.10.18 10:58, Jozef Vilcek wrote:
> >
> > So if I run 5 workers with 50 shards, I end up with:
> >
> > DurationBytes receivedRecords received
> > 2m 39s 900 MB 465,525
> > 2m 39s 1.76 GB 930,720
> > 2m 39s 789 MB 407,315
> > 2m 39s 1.32 GB 698,262
> > 2m 39s 788 MB 407,310
> >
> > Still not good but better than with 5 shards where
some workers
> did not
> > participate at all.
> > So, problem is in some layer which distributes keys /
shards
> among workers?
> >
> > On Wed, Oct 24, 2018 at 9:37 AM Reuven Lax
<re...@google.com <mailto:re...@google.com>
> <mailto:re...@google.com <mailto:re...@google.com>>
> > <mailto:re...@google.com <mailto:re...@google.com>
<mailto:re...@google.com <mailto:re...@google.com>>>> wrote:
> >
> > withNumShards(5) generates 5 random shards. It
turns out that
> > statistically when you generate 5 random shards
and you have 5
> > works, the probability is reasonably high that
some workers
> will get
> > more than one shard (and as a result not all
workers will
> > participate). Are you able to set the number of
shards larger
> than 5?
> >
> > On Wed, Oct 24, 2018 at 12:28 AM Jozef Vilcek
> <jozo.vil...@gmail.com <mailto:jozo.vil...@gmail.com>
<mailto:jozo.vil...@gmail.com <mailto:jozo.vil...@gmail.com>>
> > <mailto:jozo.vil...@gmail.com
<mailto:jozo.vil...@gmail.com>
> <mailto:jozo.vil...@gmail.com
<mailto:jozo.vil...@gmail.com>>>> wrote:
> >
> > cc (dev)
> >
> > I tried to run the example with FlinkRunner in
batch mode and
> > received again bad data spread among the workers.
> >
> > When I tried to remove number of shards for
batch mode in
> above
> > example, pipeline crashed before launch
> >
> > Caused by: java.lang.IllegalStateException:
Inputs to Flatten
> > had incompatible triggers:
> >
>
AfterWatermark.pastEndOfWindow().withEarlyFirings(AfterPane.elementCountAtLeast(40000)).withLateFirings(AfterFirst.of(Repeatedly.forever(AfterPane.elem
> > entCountAtLeast(10000)),
> >
>
Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(1
> > hour)))),
> >
>
AfterWatermark.pastEndOfWindow().withEarlyFirings(AfterPane.elementCountAtLeast(1)).withLateFirings(AfterFirst.of(Repeatedly.fo
> > rever(AfterPane.elementCountAtLeast(1)),
> >
>
Repeatedly.forever(AfterSynchronizedProcessingTime.pastFirstElementInPane())))
> >
> >
> >
> >
> >
> > On Tue, Oct 23, 2018 at 12:01 PM Jozef Vilcek
> > <jozo.vil...@gmail.com
<mailto:jozo.vil...@gmail.com> <mailto:jozo.vil...@gmail.com
<mailto:jozo.vil...@gmail.com>>
> <mailto:jozo.vil...@gmail.com
<mailto:jozo.vil...@gmail.com> <mailto:jozo.vil...@gmail.com
<mailto:jozo.vil...@gmail.com>>>> wrote:
> >
> > Hi Max,
> >
> > I forgot to mention that example is run in
streaming
> mode,
> > therefore I can not do writes without
specifying shards.
> > FileIO explicitly asks for them.
> >
> > I am not sure where the problem is.
FlinkRunner is
> only one
> > I used.
> >
> > On Tue, Oct 23, 2018 at 11:27 AM
Maximilian Michels
> > <m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>
> <mailto:m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>>> wrote:
> >
> > Hi Jozef,
> >
> > This does not look like a FlinkRunner
related
> problem,
> > but is caused by
> > the `WriteFiles` sharding logic. It
assigns keys and
> > does a Reshuffle
> > which apparently does not lead to good
data spread in
> > your case.
> >
> > Do you see the same behavior without
> `withNumShards(5)`?
> >
> > Thanks,
> > Max
> >
> > On 22.10.18 11:57, Jozef Vilcek wrote:
> > > Hello,
> > >
> > > I am having some trouble to get a
balanced
> write via
> > FileIO. Workers at
> > > the shuffle side where data per
window fire are
> > written to the
> > > filesystem receive unbalanced
number of events.
> > >
> > > Here is a naive code example:
> > >
> > > val read = KafkaIO.read()
> > > .withTopic("topic")
> > >
.withBootstrapServers("kafka1:9092")
> > >
> >
.withKeyDeserializer(classOf[ByteArrayDeserializer])
> > >
> >
> .withValueDeserializer(classOf[ByteArrayDeserializer])
> > > .withProcessingTime()
> > >
> > > pipeline
> > > .apply(read)
> > > .apply(MapElements.via(new
> > > SimpleFunction[KafkaRecord[Array[Byte],
> Array[Byte]],
> > String]() {
> > > override def apply(input:
> > KafkaRecord[Array[Byte],
> > > Array[Byte]]): String = {
> > > new
String(input.getKV.getValue,
> "UTF-8")
> > > }
> > > }))
> > >
> > >
> > >
> >
>
.apply(Window.into[String](FixedWindows.of(Duration.standardHours(1)))
> > >
> .triggering(AfterWatermark.pastEndOfWindow()
> > >
> >
> .withEarlyFirings(AfterPane.elementCountAtLeast(40000))
> > >
> >
> .withLateFirings(AfterFirst.of(Lists.newArrayList[Trigger](
> > >
> >
> Repeatedly.forever(AfterPane.elementCountAtLeast(10000)),
> > >
> > >
> >
>
Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardHours(1)))))))
> > > .discardingFiredPanes()
> > >
> >
.withAllowedLateness(Duration.standardDays(7)))
> > >
> > > .apply(FileIO.write()
> > > .via(TextIO.sink())
> > > .withNaming(new
> > SafeFileNaming(outputPath, ".txt"))
> > >
.withTempDirectory(tempLocation)
> > > .withNumShards(5))
> > >
> > >
> > > If I run this on Beam 2.6.0 with
Flink 1.5.0 on 5
> > workers (equal to
> > > number of shards), I would expect
that each worker
> > will participate on
> > > persisting shards and equally,
since code uses
> fixed
> > number of shards
> > > (and random shard assign?). But
reality is
> different
> > (see 2 attachements
> > > - statistiscs from flink task
reading from
> kafka and
> > task writing to files)
> > >
> > > What am I missing? How to achieve
balanced writes?
> > >
> > > Thanks,
> > > Jozef
> > >
> > >
> >
>
