[Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default

[ElTav]

What would you like to happen?

We're in the early stages of using Beam to run jobs on DataProc using Flink (a la the setup guide here). All of the documentation I've been able to find and the tests in interactive_runner_test.py suggest that Clusters will be auto-created and their references cached when created as part of a pipeline. However, configure_for_flink will skip caching the cluster metadata if a master url isn't already defined

I would expect the documentation to be clearer or for ClusterMetadata to be auto-cached in configure_for_flink, any chance we could make either/both of those happen? We have enterprise-specific configurations that don't jive with creating clusters ad-hoc so would prefer to simply pass along the master URL. Currently planning to set a manual entry in ie.current_env().clusters but let me know if there's another recommended approach

Snippet:

import apache_beam as beam
import apache_beam.runners.interactive.interactive_environment as ie
from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
from apache_beam.runners.portability.flink_runner import FlinkRunner
from apache_beam.options.pipeline_options import PipelineOptions

opts = {
    "region": "us-central1",
    "project_id": "project-1",
    "flink_version": "1.12",
    "flink_submit_uber_jar": True,
    "environment_type": "DOCKER",
    "environment_config": "image:tag",
    "flink_master": "000.00.00.000:8080",
}
options = PipelineOptions.from_dictionary(opts)
runner = InteractiveRunner(underlying_runner=FlinkRunner())
p = beam.Pipeline(runner=runner, options=options)
runner.configure_for_flink(p, options) # can be ib.collect/ib.show/etc
ie.current_env().clusters.__dict__

which returns

{'dataproc_cluster_managers': {},
 'master_urls': {},
 'pipelines': {},
 'default_cluster_metadata': None}

Issue Priority

Priority: 2

Issue Component

Component: beam-community

[ElTav]

cc @kevingg @rohdesamuel since it looks like you're on the right team and were Victor's intern mentors/managers based on the previous PR's I could find (1, 2)

[nika-qubit]

@ElTav Thanks for using Beam!

To clarify, the feature we added were for notebook-managed clusters. It was intentionally left out to not track any user-managed clusters.

You should still be able to use interactive runner with Flink clusters hosted by yourself (on Dataproc). Did you run into any issues? For example, networking and PCollection caching?

[ElTav]

Gotcha! Curious if you know why that decision was made?

Yeah, I've run into issues where a Pipeline isn't cached on a DeferredDataframe but I'm new to Beam + Jupyter so the error might be on my side.

collect() Runs + caches the result just fine the first time around (1-4). If I modify Equinox (e.g. to Equino) without re-instantiating the pipeline (5-6), collect tries to create a new Flink cluster instead of using the one provided in the original pipeline. It looks like the DeferredDataframe loses its references to the original pipeline but I'm not sure if something else is going on

Snippet

import apache_beam as beam
import apache_beam.runners.interactive.interactive_beam as ib
import apache_beam.runners.interactive.interactive_environment as ie
from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
from apache_beam.runners.portability.flink_runner import FlinkRunner
import os

ib.options.cache_root = "gs://internal-bucket"
opts = {
    "region": "us-central1",
    "project": "project-id",
    "staging_location": "gs://internal-bucket,
    "temp_location": "gs://internal-bucket",
    "flink_version": "1.12",
    "flink_submit_uber_jar": True,
    "environment_type": "DOCKER",
    "flink_master": "000.00.00.000:1234",
}

p = beam.Pipeline(
    InteractiveRunner(
        underlying_runner=FlinkRunner()),
        options=PipelineOptions.from_dictionary(opts)) # 1

beam_df = p | 'Read CSV' >> beam.dataframe.io.read_csv('gs://internal-bucket/solar_events.csv') # 2

filtered = beam_df[beam_df['event'].str.contains('Equinox')] # 3, 5

ib.collect(filtered) # 4, 6

[nika-qubit]

Curious if you know why that decision was made?

The decision is made so that the Beam SDK is backward compatible. It now serves users who want to use Google Cloud to host their Flink clusters implicitly (added behavior). It also retains the same behavior for users who use their own Flink clusters.

Note that you have to set ib.options.cache_root to a distributed file system such as Cloud Storage if your cluster is running remotely and distributed. You can get by with a local file system on disk if your cluster is running locally though.

collect tries to create a new Flink cluster instead of using the one provided

I'm not sure why it happens. Since you are providing the flink master, interactive runner doesn't do anything for you to manage the clusters. You are probably running into the issue because you manually added the tracking to ib.clusters. If so, please remove it because the cluster is not managed by notebooks.

[ElTav]

The decision is made so that the Beam SDK is backward compatible. It now serves users who want to use Google Cloud to host their Flink clusters implicitly (added behavior). It also retains the same behavior for users who use their own Flink clusters.

Gotcha

Note that you have to set ib.options.cache_root to a distributed file system such as Cloud Storage if your cluster is running remotely and distributed. You can get by with a local file system on disk if your cluster is running locally though.

Yup, we got it working successfully with cloud storage. The docs could use an update too since it looks like cache_dir got removed from the InteractiveRunner in 2.24.0 as far as I can tell.

I'm not sure why it happens. Since you are providing the flink master, interactive runner doesn't do anything for you to manage the clusters. You are probably running into the issue because you manually added the tracking to ib.clusters. If so, please remove it because the cluster is not managed by notebooks.

I ended up removing it. Appreciate the help regardless, closing since this is unrelated to the original issue.

[nika-qubit]

@ElTav Thanks! If needed, would you like to talk to one of our PMs to understand your use case and onboarding needs?

[aaltay]

Drive by comment: Is there an opportunity to improve the user experience? (By somehow working around the backward compatibility issue.)