stephan ewen - running flink everywhere

RunningApache Flink® Everywhere

Stephan Ewen (@StephanEwen)

How is Flink deployed?

2

Standalone Cluster Embedded Service (OSGI)

YARN SessionsStandalone Cloud

Docker on Mesos

Docker/Kubernetes

YARN->Myriad->Mesos

YARN Jobs

A two minute search on the mailing list reveals

How is Flink deployed?

3

Standalone Cluster Embedded Service (OSGI)

YARN SessionsStandalone Cloud

Docker on Mesos

Docker/Kubernetes

YARN->Myriad->Mesos

YARN Jobs

A two minute search on the mailing list reveals

Mesos Sessions

Mesos Jobs

(soon!)

How is Flink deployed?

4

Standalone Cluster Embedded Service (OSGI)

YARN SessionsStandalone Cloud

Docker on Mesos

Docker/Kubernetes

YARN->Myriad->Mesos

YARN Jobs

Users run mostly isolated jobs or multi-job sessions

Mesos Sessions

Mesos Jobs

Resource Management

5

Standalone Cluster Embedded Service (OSGI)

YARN SessionsStandalone Cloud

Docker on Mesos

Docker/Kubernetes

YARN->Myriad->Mesos

YARN Jobs

Resources controlled by the framework or another service.

Mesos Sessions

Mesos Jobs

More dimensions coming up…

6

Dynamic Resources• Number of TaskManagers

changesover job lifetime

"Trusted" processes• Run under superuser credential

and dispatch jobs

No blocking on any process type• YARN job needs to continue while

ApplicationMaster is down

Uniform vs. Heterogeneous Resources• Run different functions in

differentsize containers

• E.g., simple mapper in small container, heavy window operator in large container

Avoiding "Job Submit" step

7

Reworking the FlinkProcess Model

Flink Improvement Proposal 6

8Currently driving parties:

Core Idea• Creating composable building blocks• Create different compositions for

different scenarios

FLIP-6 design document: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=65147077

Recap: Current status (Standalone)

9

Standalone Flink Cluster

Client (2) Submit Job JobManager

TaskManager

TaskManager

TaskManager

(3) Deploy Tasks

(1) Register

Recap: Current status (YARN)

10

YARNResourceManage

r

YARN Cluster

Client(1) Submit YARN App.(FLINK)

Application Master

JobManager

TaskManager

TaskManager

TaskManager

(2) Spawn AppMaster

(4) StartTaskManagers

(8) DeployTasks

(3) Poll status

(6) AllTaskManagerstarted

(5) Register(7) Submit

Job

The Building Blocks

11

• ClusterManager-specific• May live across jobs• Manages available

Containers/TaskManagers• Used to acquire / release resources

ResourceManager

TaskManagerJobManager

• Registers at ResourceManager• Gets tasks from one or more

JobManagers

• Single job only, started per job• Thinks in terms of "task slots"• Deploys and monitors job/task execution

Dispatcher

• Lives across jobs• Touch-point for job submissions• Spawns JobManagers• May spawn ResourceManager

The Building Blocks

12

ResourceManager

(1) Request slots

TaskManager

JobManager

(2) Start TaskManager

(3) Register

(4) Deploy Tasks

Building Flink-on-YARN

13

YARNResourceManage

r

YARN Cluster

YARN Cluster Client

(1) Submit YARN App.(JobGraph / JARs)

Application MasterFlink-YARN

ResourceManager

JobManager TaskManager

TaskManager

TaskManager

(2) Spawn AppMaster

(4) StartTaskManagers

(6) DeployTasks

(5) Register(3) Request slots

Building Flink-on-YARNMain differences from current YARN mode All containers started with JARs, config files in classpath Credentials & Secrets are strictly bound to a single job Slots are allocated/released as needed/freed

• Basic building block for elastic resource usage

Client disconnects after submitting job, does not need to wait until TaskManagers are up

14

Building Flink-on-YARN (separate RM)

15

YARNResourceManage

r

YARN Cluster

YARN Cluster Client

(1) Submit YARN App.(JobGraph / JARs)

Application MasterFlink-YARN

ResourceManager

JobManager TaskManager

TaskManager

TaskManager

(2) Spawn AppMaster

(4) StartTaskManagers

(6) DeployTasks

(5) Register

(4) Requestslots

(3) StartJobMngr

Building Flink-on-YARN (w/ dispatcher)

16

YARNResourceManage

r

YARN Cluster

YARN Cluster Client

(1) HTTP POST JobGraph/Jars

Application MasterFlink-YARN

ResourceManager

JobManager TaskManager

TaskManager

TaskManager

(3) Spawn AppMaster

(5) StartTaskManagers

(7) DeployTasks

(6) Register

Flink YARNDispatcher

(2) Submit YARN App.(JobGraph / JARs)

(4) Request slots

Building Flink-on-Mesos

17

Mesos Master

Mesos Cluster

Mesos Cluster Client

(1) HTTP POST JobGraph/Jars

Flink Master ProcessFlink Mesos

ResourceManager

JobManager TaskManager

TaskManager

TaskManager

(3) Start Process (and supervise)

(5) StartTaskManagers

(7) DeployTasks

(6) Register(4) Request slots

Flink MesosDispatcher

(2) Allocate containerfor Flink master

Building Standalone

18Standalone Cluster

Flink Cluster Client

(1) Submit JobGraph/Jars

Flink Master ProcessStandalone

ResourceManager

TaskManager

TaskManager

TaskManager(7) Deploy Tasks

(1) Register(3) Request

slotsJobManager

JobManager

Dispatcher

(2) Start JobMngr

Standby Master Process

Standby Master Process

Master Container

Flink Master Process

Building Flink-on-Docker/K8S

19

Flink-ContainerResourceManage

r

JobManager

Program Runner

(2) Run & Start

Worker ContainerTaskManager

Worker ContainerTaskManager

Worker ContainerTaskManager

(3) Register

(1) Container framework starts Master & Worker Containers

(4) Deploy Tasks

Building Flink-on-Docker/K8S This is a blueprint for all setups where external services

control resources and start new TaskManagers• For example AWS EC2 Flink image with auto-scaling groups

Can be extended to have N equal containers, out of which one becomes master, remainder workers

With upcoming dynamic-scaling feature (see Till's talk), JobManager scales job to use all available resources

20

21

Multi-Job Sessions

Example: YARN session

ApplicationMaster

Flink-YARNResourceManager

(5) Request

slotsJobManager

(A)

JobManager

(B)

Dispatcher

(4) Start

JobMngr

YARNResourceManage

r

YARN Cluster

Client

(1) Submit YARN App.(FLINK – session)

TaskManager

TaskManager

TaskManager

(2) Spawn AppMaster

(6) StartTaskManagers

(8, 12) Deploy Tasks

(7) Register

(3) Submit Job A

(11) Request slots

(10) StartJobMngr

(9) SubmitJob B

22

Sessions vs. Jobs For each Job submitted, the session will spawn its own

JobManager

All jobs run under session-user credentials

ResourceManager holds on to containers for a certain time• Jobs quickly following one another reuse containers (quicker response)

Internally, sessions build on the dispatcher component

23

24

Wrap-up

More stuff Dynamically acquire/release resources

• Slots are allocated/released from Resource Manager as needed• ResourceManager allocates/releases containers over time• Strong interplay with "Dynamic Scaling" (rf. talk by Till yesterday)

Resource Profiles: Containers of different size• Requests can pass a "profile" (CPU / memory / disk), or simply use

"default profile"• Resource Managers YARN & Mesos can allocate respective containers

25

Wrapping it up It’s a zoo of cluster managers out there

• Following different paradigms

Usage patterns vary because of Flink's broad use cases• Isolated long running jobs vs. many short-lived jobs• Shared clusters vs. per-user authenticated resources

We are making "jobs" and "sessions" explicit constructs Flexible building blocks, composed in various ways to

accommodate different scenarios26

27

Appendix

Flink Streaming cornerstones

28

Low latency

High Throughput

Well-behavedflow control

(back pressure)

Make more sense of data

Works on real-timeand historic data

PerformantStreaming

Event Time

APIsLibraries

StatefulStreaming

Globally consistentsavepoints

Exactly-once semanticsfor fault tolerance

Windows &user-defined state

Flexible windows(time, count, session, roll-your own)

Complex Event Processing