Splittable DoFn

ALEX VAN BOXEL

09 – 24 – 2018

Alex Van Boxel

@alexvb

• Day job: Big Data Architect at Vente-Exclusive.com. Part of the Vente-Privee Group. Specialised in everything Google
• Nightlife: Having fun reading documentation and stalking Google Product Managers as a Google Cloud Expert
• Needed an excuse to use this slide in the deck because it’s so cool

Agenda

• Who needs agenda’s. It’s always full of meetings...

Why a new input?

Entities�Cloud Pub/Sub

Stream -

BT

Why a new input?

Stream -

BT

It all began in ...

Happy

not to mess with Unbounded IO Connector

Learning

Splittable DoFn

Documentation

• Apache Beam Blog
• Design Doc
• Java Doc

Splittable Support

What

is a SDF?

What can’t a DoFn do?

• Work inside a DoFn can’t be parallelised further
• Can’t produce infinite output

​

What can a SDF do?

• Split work in parts though restrictions (yes: parallelize)
• Ways of checkpointing, resuming
• Infinite output

SDF has super powers

• It takes an input element
• Is composable

​

Example: File on storage

• Input: storage URL
• Restriction: [0,object.size)

​

Example: BigTable

• Input: table
• Restriction: [key.start,key.end)

​

Idea: Infinite storage stream

• SDF1^u: watch a storage bucket for change
• SDF2^b: object url as input, read the object

​

@

DoFn.BoundedPerElement

@

DoFn.GetInitialRestriction

• Use this to set the range for the given input element
• OffsetRange is a good general starting point

Define the range

@GetInitialRestriction�public OffsetRange getInitialRestriction(String url) {� Long size = sizeOf(url) � return new OffsetRange(0, size);�}�

@

DoFn.SplitRestriction

• Optional: Only called once
• After that it’s up to the tracker

First split, you know best

@SplitRestriction�public void splitRestriction(String element, OffsetRange restriction,� OutputReceiver<OffsetRange> ranges) {� long x2 = restriction.getTo();� ranges.output(new OffsetRange(restriction.getFrom(), x2 / 2));� ranges.output(new OffsetRange(x2 / 2, x2));�}�

@

DoFn.NewTracker

• Optional, ranges can have a default tracker

​

Get ready to work

@ProcessElement�public void process(ProcessContext c, OffsetRangeTracker tracker) {� OffsetRange currentRestriction = tracker.currentRestriction();�

…�}

​

Get ready to work

long pos = currentRestriction.getFrom();� pos = nextValidPosition(pos);

for (; tracker.tryClaim(pos); ) {� // no need to claim every pos� for( block ) {�� c.output(the_work);� }�

Get ready to work

long pos = currentRestriction.getFrom();� pos = nextValidPosition(pos);

for (; tracker.tryClaim(pos); ) {�� for( block ) {� // you can do work for pos past the currentRestriction.getTo()� c.output(the_work);� }�

Bounded

Example code:BoundedOverlapFn.java

@

DoFn.UnboundedPerElement

Define the range

@GetInitialRestriction�public OffsetRange getInitialRestriction(String url) {� Long begin = begin(url) � return new OffsetRange(begin, Long.MAX_VALUE);�}�

DoFn.ProcessContinuation

• Unbounded work needs more flow control

​

Unbound flow control

​

return DoFn.ProcessContinuation.stop()

​

Unbound flow control

​

​

return DoFn.ProcessContinuation.resume()

​

Unbound flow control

​

​

return DoFn.ProcessContinuation.resume()

.withResumeDelay(Duration.millis(100))

​

Claim/Resume gotcha

​

• Resume after claim -> position++
• Resume without claim -> position = position

Unbounded

Example code:SecondTickSplitFn.java

So what

happened to the Mongo Changestream use case?

Conclusion

• Remember… composable, your imagination is the limit
• How do I split up my work? OffsetRange is a good starter
• Play: github.com/alexvanboxel/beam-demo/

Questions & (Answers)

• I prefer blue police boxes… When is Doctor Who starting again?

Thank You