Re: GCS staging location when uploading artifacts from python for dataflow

[Luke Cwik <lc...@google.com>]

Google has been trying to migrate away from Dataflow implementation for
Python for staging which isn't easy to change onto the artifact service and
provisioning service with Dataflow prime. With these the container can
choose what to do with the resources and should have control over how
certain artifacts are treated specially so it might work out of the box
with Dataflow prime for you and if not you can control what the Python
container is doing.

On Mon, Dec 13, 2021 at 12:28 PM Steve Niemitz <sn...@apache.org> wrote:

> I spent a bunch of time on this over the last week and arrived at
> something that works.  I had to try a few different ways to make this work
> though.
>
> 1) Simply removing the unique job prefix and appending the sha256 to the
> uploaded artifact fails because some paths are specifically hard coded
> (dataflow_python_sdk.tar, pipeline.pb, etc) in the runtime stager (which
> seems to be closed-source in dataflow from what I can tell).
>
> 2) If instead I prefix the artifacts with their sha256 as a path (eg
> gs://my-bucket/<sha256>/dataflow_python_sdk.tar) the job fails again at
> runtime, with the stager complaining that the directory structure isn't
> "flat", it seems like it doesn't support artifacts from multiple paths.
>
> I arrived at a solution that does work, but isn't the most optimal.  I
> first stage the artifacts to the same bucket, but in a different path (eg
> gs://my-bucket/cache/<sha256>), these are only staged if they don't already
> exist.  Then, I can use a GCS copy operation to copy them from the cache
> directory to the "unique" staging directory.  The copy operation is
> ~instant and doesn't require the user to download/reupload the artifact
> again.
>
> I think ideally we'd fix #1 to work, it seems like the worker bootstrap
> just needs to be updated to be more robust (although maybe there's more
> issues there?).  In any case, what kind of appetite would there be for a PR
> for this?  It could be additionally gated behind a pipeline option if we
> don't want it to be the default.
>
> On Tue, Dec 7, 2021 at 12:49 PM Steve Niemitz <sn...@apache.org> wrote:
>
>> I noticed that the python dataflow runner appends some "uniqueness" (the
>> timestamp) [1] to the staging directory when staging artifacts for a
>> dataflow job.  This is very suboptimal because it makes caching artifacts
>> between job runs useless.
>>
>> The jvm runner doesn't do this, is there a good reason the python one
>> does?  Or is this just an oversight that hasn't been fixed yet?
>>
>> [1]
>> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py#L467
>>
>

Re: GCS staging location when uploading artifacts from python for dataflow

[Luke Cwik <lc...@google.com>]

On Sat, Jan 15, 2022 at 7:01 AM Steve Niemitz <sn...@apache.org> wrote:

> That certainly makes sense.  Would the flow look like:
> job submission stages to GCS/rewrites artifacts in the proto to point at
> GCS -> job requests artifact from service (via the same gs:// url) ->
> service downloads from GCS and stream artifact back?
>
> That is the gist of it.


> One thing I find a little confusing with the artifact process in general
> is that (in what you're describing) it seems like there's actually two
> artifact services involved.  One is used when the job is being staged, and
> one when the job is running, but (at least in our case) these are two
> different services running in different places, with different access
> restrictions.
>

To the user it shouldn't matter whether there is one or two instances. The
responsibilities are split where the first part is during job creation and
is responsible for creating the artifacts before the job is running. The
other responsibility is about ensuring the worker has access to all the
artifacts that were there.


>
> For example, our internal expansion service returns artifacts that point
> at our internal blobstore (URL resources).  The way things work now in the
> staging process is that the internal artifact service is never even used,
> the stager downloads the artifacts directly from the URL and uploads them
> to GCS.  This is (luckily) how things work now by default, but are also
> required because the job running in dataflow couldn't access the same
> resources on the internal blobstore.  Is the assumption that the dataflow
> runner would always do something like this (convert all resources to URL
> resources pointing to GCS)?
>

Yes this is how it works today but Dataflow will transition to using the
artifact service for staging during job creation and for retrieval when the
job is running. The migration is not high on the priority list so it will
likely take a while before it starts and in the future Dataflow may choose
a different way to store the artifacts instead of GCS.


>
> Is there a design doc I can read about how this whole process should work
> in the ideal world?
>
>
There are
*
https://docs.google.com/document/d/12zNk3O2nhTB8Zmxw5U78qXrvlk5r42X8tqF248IDlpI/edit

*
https://docs.google.com/document/d/1XaiNekAY2sptuQRIXpjGAyaYdSc-wlJ-VKjl04c8N48/edit
* https://s.apache.org/beam-fn-api-container-contract

Feel free to comment on them.


> On Fri, Jan 14, 2022 at 6:58 PM Luke Cwik <lc...@google.com> wrote:
>
>> Google has been trying to migrate away from Dataflow implementation for
>> Python for staging which isn't easy to change onto the artifact service and
>> provisioning service with Dataflow prime. With these the container can
>> choose what to do with the resources and should have control over how
>> certain artifacts are treated specially so it might work out of the box
>> with Dataflow prime for you and if not you can control what the Python
>> container is doing.
>>
>> On Mon, Dec 13, 2021 at 12:28 PM Steve Niemitz <sn...@apache.org>
>> wrote:
>>
>>> I spent a bunch of time on this over the last week and arrived at
>>> something that works.  I had to try a few different ways to make this work
>>> though.
>>>
>>> 1) Simply removing the unique job prefix and appending the sha256 to the
>>> uploaded artifact fails because some paths are specifically hard coded
>>> (dataflow_python_sdk.tar, pipeline.pb, etc) in the runtime stager (which
>>> seems to be closed-source in dataflow from what I can tell).
>>>
>>> 2) If instead I prefix the artifacts with their sha256 as a path (eg
>>> gs://my-bucket/<sha256>/dataflow_python_sdk.tar) the job fails again at
>>> runtime, with the stager complaining that the directory structure isn't
>>> "flat", it seems like it doesn't support artifacts from multiple paths.
>>>
>>> I arrived at a solution that does work, but isn't the most optimal.  I
>>> first stage the artifacts to the same bucket, but in a different path (eg
>>> gs://my-bucket/cache/<sha256>), these are only staged if they don't already
>>> exist.  Then, I can use a GCS copy operation to copy them from the cache
>>> directory to the "unique" staging directory.  The copy operation is
>>> ~instant and doesn't require the user to download/reupload the artifact
>>> again.
>>>
>>> I think ideally we'd fix #1 to work, it seems like the worker bootstrap
>>> just needs to be updated to be more robust (although maybe there's more
>>> issues there?).  In any case, what kind of appetite would there be for a PR
>>> for this?  It could be additionally gated behind a pipeline option if we
>>> don't want it to be the default.
>>>
>>> On Tue, Dec 7, 2021 at 12:49 PM Steve Niemitz <sn...@apache.org>
>>> wrote:
>>>
>>>> I noticed that the python dataflow runner appends some "uniqueness"
>>>> (the timestamp) [1] to the staging directory when staging artifacts for a
>>>> dataflow job.  This is very suboptimal because it makes caching artifacts
>>>> between job runs useless.
>>>>
>>>> The jvm runner doesn't do this, is there a good reason the python one
>>>> does?  Or is this just an oversight that hasn't been fixed yet?
>>>>
>>>> [1]
>>>> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py#L467
>>>>
>>>

Re: GCS staging location when uploading artifacts from python for dataflow

[Steve Niemitz <sn...@apache.org>]

That certainly makes sense.  Would the flow look like:
job submission stages to GCS/rewrites artifacts in the proto to point at
GCS -> job requests artifact from service (via the same gs:// url) ->
service downloads from GCS and stream artifact back?

One thing I find a little confusing with the artifact process in general is
that (in what you're describing) it seems like there's actually two
artifact services involved.  One is used when the job is being staged, and
one when the job is running, but (at least in our case) these are two
different services running in different places, with different access
restrictions.

For example, our internal expansion service returns artifacts that point at
our internal blobstore (URL resources).  The way things work now in the
staging process is that the internal artifact service is never even used,
the stager downloads the artifacts directly from the URL and uploads them
to GCS.  This is (luckily) how things work now by default, but are also
required because the job running in dataflow couldn't access the same
resources on the internal blobstore.  Is the assumption that the dataflow
runner would always do something like this (convert all resources to URL
resources pointing to GCS)?

Is there a design doc I can read about how this whole process should work
in the ideal world?

On Fri, Jan 14, 2022 at 6:58 PM Luke Cwik <lc...@google.com> wrote:

> Google has been trying to migrate away from Dataflow implementation for
> Python for staging which isn't easy to change onto the artifact service and
> provisioning service with Dataflow prime. With these the container can
> choose what to do with the resources and should have control over how
> certain artifacts are treated specially so it might work out of the box
> with Dataflow prime for you and if not you can control what the Python
> container is doing.
>
> On Mon, Dec 13, 2021 at 12:28 PM Steve Niemitz <sn...@apache.org>
> wrote:
>
>> I spent a bunch of time on this over the last week and arrived at
>> something that works.  I had to try a few different ways to make this work
>> though.
>>
>> 1) Simply removing the unique job prefix and appending the sha256 to the
>> uploaded artifact fails because some paths are specifically hard coded
>> (dataflow_python_sdk.tar, pipeline.pb, etc) in the runtime stager (which
>> seems to be closed-source in dataflow from what I can tell).
>>
>> 2) If instead I prefix the artifacts with their sha256 as a path (eg
>> gs://my-bucket/<sha256>/dataflow_python_sdk.tar) the job fails again at
>> runtime, with the stager complaining that the directory structure isn't
>> "flat", it seems like it doesn't support artifacts from multiple paths.
>>
>> I arrived at a solution that does work, but isn't the most optimal.  I
>> first stage the artifacts to the same bucket, but in a different path (eg
>> gs://my-bucket/cache/<sha256>), these are only staged if they don't already
>> exist.  Then, I can use a GCS copy operation to copy them from the cache
>> directory to the "unique" staging directory.  The copy operation is
>> ~instant and doesn't require the user to download/reupload the artifact
>> again.
>>
>> I think ideally we'd fix #1 to work, it seems like the worker bootstrap
>> just needs to be updated to be more robust (although maybe there's more
>> issues there?).  In any case, what kind of appetite would there be for a PR
>> for this?  It could be additionally gated behind a pipeline option if we
>> don't want it to be the default.
>>
>> On Tue, Dec 7, 2021 at 12:49 PM Steve Niemitz <sn...@apache.org>
>> wrote:
>>
>>> I noticed that the python dataflow runner appends some "uniqueness" (the
>>> timestamp) [1] to the staging directory when staging artifacts for a
>>> dataflow job.  This is very suboptimal because it makes caching artifacts
>>> between job runs useless.
>>>
>>> The jvm runner doesn't do this, is there a good reason the python one
>>> does?  Or is this just an oversight that hasn't been fixed yet?
>>>
>>> [1]
>>> https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py#L467
>>>
>>