You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/11/14 20:52:50 UTC
[GitHub] [beam] ElTav opened a new issue, #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
ElTav opened a new issue, #24161:
URL: https://github.com/apache/beam/issues/24161
### 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](https://cloud.google.com/dataproc/docs/concepts/components/flink)). All of the documentation I've been able to find and the tests in [`interactive_runner_test.py`](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/interactive/interactive_runner_test.py#L505-L549) suggest that Clusters will be auto-created and their references cached when created as part of a pipeline. However, `configure_for_flink` will [skip](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/interactive/interactive_runner.py#L224-L259) 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ElTav commented on issue #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
Posted by GitBox <gi...@apache.org>.
ElTav commented on issue #24161:
URL: https://github.com/apache/beam/issues/24161#issuecomment-1315666959
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
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] aaltay commented on issue #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
Posted by GitBox <gi...@apache.org>.
aaltay commented on issue #24161:
URL: https://github.com/apache/beam/issues/24161#issuecomment-1317892400
Drive by comment: Is there an opportunity to improve the user experience? (By somehow working around the backward compatibility issue.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] aaltay commented on issue #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
Posted by GitBox <gi...@apache.org>.
aaltay commented on issue #24161:
URL: https://github.com/apache/beam/issues/24161#issuecomment-1319230647
> > Drive by comment: Is there an opportunity to improve the user experience? (By somehow working around the backward compatibility issue.)
>
> The "backward compatibility" is basically the feature brings added value to serve notebook-managed clusters while the user's use case is already supported by the SDK.
>
> An AI I can come up with is:
>
> * We either need to figure out how the onboarding story ended up in this situation;
>
> * We are engaging the user for clarification now...
> * Or have to make some adjustments in code/docstrings for advanced users who explore features by reading through source code
>
> * We can emphasize in the code that it's for notebook-managed clusters (I'll send a PR to do it)
> * We can add helpers in the SDK to cater user-managed resources in an interactive setup (@liferoad this can be a feature of your proposed easy-to-use Python layer)
> * It'll be good to have some notion in the SDK to separate pipeline building, running and runtime environment configuration (this falls into the category)
Thank you. IMO, just having some docs might be sufficient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ElTav commented on issue #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
Posted by GitBox <gi...@apache.org>.
ElTav commented on issue #24161:
URL: https://github.com/apache/beam/issues/24161#issuecomment-1315803390
>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](https://github.com/apache/beam/tree/master/sdks/python/apache_beam/runners/interactive#portability-across-storage-systems) 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ElTav closed issue #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
Posted by GitBox <gi...@apache.org>.
ElTav closed issue #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
URL: https://github.com/apache/beam/issues/24161
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] KevinGG commented on issue #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
Posted by GitBox <gi...@apache.org>.
KevinGG commented on issue #24161:
URL: https://github.com/apache/beam/issues/24161#issuecomment-1319015509
> Drive by comment: Is there an opportunity to improve the user experience? (By somehow working around the backward compatibility issue.)
The "backward compatibility" is basically the feature brings added value to serve notebook-managed clusters while the user's use case is already supported by the SDK.
An AI I can come up with is:
- We either need to figure out how the onboarding story ended up in this situation;
- We are engaging the user for clarification now...
- Or have to make some adjustments in code/docstrings for advanced users who explore features by reading through source code
- We can emphasize in the code that it's for notebook-managed clusters (I'll send a PR to do it)
- We can add helpers in the SDK to cater user-managed resources in an interactive setup (@liferoad this can be a feature of your proposed easy-to-use Python layer)
- It'll be good to have some notion in the SDK to separate pipeline building, running and runtime environment configuration (this falls into the category)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] KevinGG commented on issue #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
Posted by GitBox <gi...@apache.org>.
KevinGG commented on issue #24161:
URL: https://github.com/apache/beam/issues/24161#issuecomment-1315919364
@ElTav Thanks! If needed, would you like to talk to one of our PMs to understand your use case and onboarding needs?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] KevinGG commented on issue #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
Posted by GitBox <gi...@apache.org>.
KevinGG commented on issue #24161:
URL: https://github.com/apache/beam/issues/24161#issuecomment-1315796324
> 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ElTav commented on issue #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
Posted by GitBox <gi...@apache.org>.
ElTav commented on issue #24161:
URL: https://github.com/apache/beam/issues/24161#issuecomment-1314367466
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](https://github.com/apache/beam/pull/16904), [2](https://github.com/apache/beam/pull/16741))
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] KevinGG commented on issue #24161: [Feature Request]: Python SDK InteractiveRunner/FlinkRunner should cache clusters with Master URLs by default
Posted by GitBox <gi...@apache.org>.
KevinGG commented on issue #24161:
URL: https://github.com/apache/beam/issues/24161#issuecomment-1314435986
@ElTav Thanks for using Beam!
To clarify, the feature we added were for [notebook-managed clusters](https://cloud.google.com/dataflow/docs/guides/interactive-pipeline-development#interactive_flinkrunner_on_notebook-managed_clusters). It was intentionally [left out](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/interactive/interactive_runner.py#L260-L261) 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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org