Hi Zhanghao, thank you for the proposal, it is very interesting, especially for teams with only Ray clusters.
I spent some time with AI's help spinning up the proposed architecture and running some experiments and alternatives, I was not however able to determine if these same alternatives were assessed and wanted to get your feedback. Alternative A: Ray contributes only TaskManagers to an existing Flink cluster In this variant, the JobManager (and any CPU-only TaskManagers) stay wherever Flink already runs: Kubernetes, YARN, or a managed service. Ray actors start extra TaskManagers that register with that JobManager and advertise GPUs as an external resource (external-resource.gpu.amount). GPU operators are routed to them through slot sharing groups with fine-grained resource requirements. No RayResourceManagerDriver is needed, and Ray never owns the JobManager. - Was this "Ray as a TaskManager pool" model considered? If so, why was it rejected? - Several of the open lifecycle questions (JobManager restart, recovering TaskManager handles, fencing stale requests, HA) seem to disappear if Ray never hosts the JobManager. Is there a capability in the FLIP that only works when the JobManager runs on Ray? - In my prototype Flink placed CPU-only operators on the GPU TaskManagers, because nothing keeps ordinary work off GPU capacity. Does the FLIP intend to address this, for example by sizing GPU TaskManagers to the GPU slot sharing group exactly, or with some tag/affinity mechanism? Or is it out of scope? Alternative B: Ray schedules the Python UDF workers, not the TaskManager Here the TaskManager doesn't run on Ray and requests no GPU resources. When the TaskManager starts a Python UDF worker, that worker is launched as a Ray task with its own GPU. This matches the pen Dian described. In my prototype two Python workers in one TaskManager each got a distinct GPU from Ray. I did this without Flink changes, by swapping the Python executable PyFlink launches for a small shim wrapper. - Was allocating GPUs per Python worker considered as the primary model, rather than per TaskManager? For GPU inference, the Python worker seems to be the actual GPU consumer.If the follow-up "PyFlink with Ray" FLIP is going to touch the Python world it make sense to treat per-worker GPU allocation as the main path andTaskManager-level GPUs as a fallback? - Would the community be open to a pluggable way of launching Python workers, for example something similar to Beam's external worker environment, so the Python worker can be hosted by Ray (or something else). Thanks for your consideration and effort on the proposal! On Wed, Oct 7, 2026 at 10:59 PM Dian Fu <[email protected]> wrote: > Hi Zhanghao, > > Thanks for driving this FLIP. This is an interesting topic. > > Ray has become an important execution layer in the Python data and AI > ecosystem. Besides Ray-native libraries, there are quite a few engines > such as Daft and Modin that use Ray as an execution backend, and Dask > workloads can also run through Dask-on-Ray. Supporting Flink on Ray > would allow PyFlink users to reuse an existing Python-native cluster > and share Ray’s CPU, GPU, and resource scheduling capabilities. > > The overall architecture looks reasonable: use Ray actors as > deployment units that supervise Flink JVM processes, while relying on > Ray for resource allocation and keeping scheduling, checkpointing, and > task recovery in the existing Flink runtime. > > I have several questions about the design details. > > ### TaskManager actor creation and teardown > > The FLIP could further clarify how TaskManager actor creation and > ownership work across a JobManager failover, for example: > > - Whether a JobManager failover causes existing TaskManager actors to > restart. My expectation is that healthy TaskManager actors and their > JVM processes should remain running. > - How a recovered JobManager obtains handles to existing TaskManager > actors. > - How duplicate TaskManager actor creation is prevented during failover. > - Whether TaskManager actors are named (if so, how the name will be > derived) and detached from their creator. > - How stale worker-management requests from the previous JobManager > are fenced after a failover? > - How detached TaskManager actors are cleaned up after the application > reaches a terminal state. > > The design should also distinguish a JobManager restart from an > application shutdown. Restarting or replacing the JobManager should > preserve healthy TaskManager actors, while application shutdown should > explicitly and reliably terminate all detached TaskManager actors. > > ### High availability model > > Whether the proposed deployment reuses an existing Flink HA > implementation or will introduce a new `RayHighAvailabilityServices`? > > Besides, whether JobManager HA supports the following cases: > - one restartable JobManager actor; or > - multiple active/standby JobManager actors. > > ### GPU resource management > > The FLIP could clarify the GPU allocation and isolation granularity. > > When a TaskManager actor requests GPU resources, Ray sets > `CUDA_VISIBLE_DEVICES` for the actor. The TaskManager JVM and the > Python UDF workers launched by it may therefore see the same set of > GPUs. If multiple GPU-consuming Python workers run concurrently in the > same TaskManager, they may compete for GPU memory and compute > resources. > > Configuring one slot per GPU-enabled TaskManager can reduce the > likelihood of such contention, but it does not by itself guarantee > isolation. A Flink slot may contain multiple operators or operator > chains, and an operator chain may contain multiple GPU-consuming UDFs. > > Supporting multiple independently isolated GPU consumers in one > TaskManager would require an additional per-slot or per-Python-worker > device allocation mechanism, including assigning a distinct > `CUDA_VISIBLE_DEVICES` value when each Python worker is launched. The > behavior should be clearly defined. For the initial version, the FLIP > could also document that GPU allocation and visibility are defined at > the TaskManager actor level and that no additional slot- or UDF-level > isolation is provided. > > ### API and naming > > The FLIP currently uses both `init_flink()` and `init_flink_on_ray()`. > I guess it could be simplified as following: > > import pyflink.ray as flink_on_ray > flink_on_ray.init(...) > > The FLIP could also clarify whether `init()` invokes `ray.init()` > automatically, how it behaves when a Ray context already exists, and > whether the corresponding shutdown operation terminates only the Flink > application or the entire Ray cluster. > > The actor names could also be made consistent. What about renaming > `RayFlinkJobManagerActor` to `RayJobManagerActor` so that it is > symmetric with `RayTaskManagerActor`. > > Thanks, > Dian > > On Fri, Sep 18, 2026 at 12:50 PM Zhanghao Chen > <[email protected]> wrote: > > > > Hi all, > > I would like to start a discussion about FLIP-608: Flink on Ray — Ray > Resource Backend For Flink [1]. > > This FLIP proposes a new Ray resource backend to allow running Flink > jobs natively on Ray. The hybrid approach takes the best of both worlds > —battle-tested streaming and batch unific computing plus heterogeneous > scheduling and the Ray AI eco-system (Data/Serve/Train). One can therefore > setup the entire real-time AI workflow in a single cluster with higher > resource efficiency and lower maintenance cost. > > The proposal focuses on supporting running Flink jobs on Ray in > Application/Session mode by translating Flink resource requirements to Ray. > The Flink core mechanisms like RPC, shuffle, checkpointing, failover and > slot management are preserved. > > A follow-up FLIP will focus on the interoperability between PyFlink and > Ray by supporting conversion between Flink Dataframe/Datastream/Table and > Ray Dataset. > > Looking forward to your feedback! > > [1] > https://cwiki.apache.org/confluence/spaces/FLINK/pages/449286348/FLIP-608+Flink+on+Ray+%E2%80%94+Ray+Resource+Backend+For+Flink > > > > Best, > > Zhanghao Chen >
