[Feature Request]: Allow a flag so that "with pipeline" will not wait_until_finish so that this paradigm can be used for streaming jobs

[lazarillo]

What would you like to happen?

Currently, using the Python SDK, if the preferred with beam.Pipeline() as pipeline: context manager is used, a streaming job will never finish and will hang the task.

I am using Apache Beam within a CI/CD pipeline on Google's Dataflow. I want the deploy agent to launch the job and then complete. Then I can use the logs that I am receiving and Dataflow's GUI to monitor the job. Leaving the agent hung provides no value. For now, I am not using the context menu, but I prefer to use it as a best practice, especially since __exit()__ does more than simply wait_until_finish(), and as Beam progresses, it will likely continue to do more.

However, I understand the need for the CLI job to "hang" in many instances. Therefore, maybe an option can be added to StandardOptions to flag a job as "releaseable" (I'm sure there is a better flag name), then a small if clause can be placed here to only wait_until_finish when the flag is not set.

I am happy to help with this implementation, assuming the core team agrees with this approach.

Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[liferoad]

For your use case, you should not use with beam.Pipeline() as pipeline. Just do p = beam.Pipeline() and later do p.run().

[damccorm]

Yeah, +1 to what @liferoad said. In general, the with ... syntax is used when you have some cleanup that you want to do at the end of execution, which implies that everything has completed in the with block. p = beam.Pipeline() and p.run() is the right way to do this.

I'm going to close here, feel free to comment or reopen if you have further questions

[lazarillo]

Yes, I understand that I can do it the way that you all mention. But there are two problems:

1. Context managers are "the Pythonic way" because as users, we assume that __enter__() and __exit()__ are doing useful, hidden things that should be best practices. For instance, file or DB connection handles are gracefully and automatically closed, no matter what happens within the manager.
2. There is no documentation anywhere, at least nowhere that I could find, that explicitly states that one should not use the context manager when working with streaming jobs. I just discovered this after some hunting + guessing. Minimally, there should be some documentation stating that a context manager cannot/should not be used for streaming.
   • I don't have evidence, but I would expect that a lot of Apache Beam users are using it for streaming purposes. Otherwise, we can just use Spark, which has been around longer, also has good documentation, and has a wider base of users with experience on the platform.

I also wonder, what is the purpose of using the context manager, if the __enter__() and __exit__() are not needed? Or if they are needed, then why isn't a context manager supported for streaming?

I doubt I have the expertise to help with the implementation, but if you would like me to work on documentation stating that context managers should not be used for streaming, I can do that. I would just like some promise / confidence that the PR I create would be merged in (assuming it's good).

[damccorm]

Context managers are "the Pythonic way" because as users, we assume that enter() and exit() are doing useful, hidden things that should be best practices. For instance, file or DB connection handles are gracefully and automatically closed, no matter what happens within the manager.

I'd say that is one purpose, the other purpose is to clearly encapsulate usage of resources. It is generally expected that when you enter a with Foo as bar: block that you will stay in the block while bar is still an active reference to Foo's resources. The __exit()__ best practices you mention are usually making sure that you've actually released your references because you're no longer holding on to the resource. That is basically what this means for Beam

There is no documentation anywhere, at least nowhere that I could find, that explicitly states that one should not use the context manager when working with streaming jobs.

+1 on adding docs, I agree this could be clearer.

I don't have evidence, but I would expect that a lot of Apache Beam users are using it for streaming purposes.

This is definitely true :)

context managers should not be used for streaming

This isn't quite right; context managers shouldn't be used for pipelines that will never terminate. It is possible for streaming pipelines to be launched and still eventually finish (either because they were written in a way where the watermark can advance to the end of the global window or because a user cancels them).

if you would like me to work on documentation stating that context managers should not be used for streaming, I can do that. I would just like some promise / confidence that the PR I create would be merged in (assuming it's good).

Sure, always happy to review a doc improvement

[lazarillo]

+1 on adding docs, I agree this could be clearer.

OK, I'll try to work on it this weekend.

context managers should not be used for streaming

This isn't quite right; context managers shouldn't be used for pipelines that will never terminate. It is possible for streaming pipelines to be launched and still eventually finish (either because they were written in a way where the watermark can advance to the end of the global window or because a user cancels them).

if you would like me to work on documentation stating that context managers should not be used for streaming, I can do that. I would just like some promise / confidence that the PR I create would be merged in (assuming it's good).

Sure, always happy to review a doc improvement

Please help me understand this a bit more, so that I can write the doc better. I don't understand this case where a watermark can advance to the end of the global window.

Regardless, I will stress the fact that it will "hang" and not release the job, which might be fine for some cases, but would not be good for a CI/CD pipeline to launch a job.

I'll also look through the context manager a bit more, to better understand what it is doing and what it is not doing.

[damccorm]

Please help me understand this a bit more, so that I can write the doc better. I don't understand this case where a watermark can advance to the end of the global window.

Basically any source (streaming or batch) can say that it is done performing work by advancing the watermark to the end of the global window (https://beam.apache.org/documentation/programming-guide/#watermark-estimation). At that point (and once the in flight data is processed), even if the pipeline is launched in streaming mode it will complete.

[langmore]

Some documentation would be nice. Echoing what @lazarillo has said, it's hard to have confidence that pipeline.run() and wait_until_finished will do as expected if I can't point to documentation. The worry is that enter/exit are doing more.