On Tue, Jun 29, 2021 at 2:39 PM Jan Lukavský <je...@seznam.cz
<mailto:je...@seznam.cz>> wrote:
On 6/29/21 11:04 PM, Robert Bradshaw wrote:
> You can configure the environment in the current state, you
just have
> to run your own expansion service that has a different
environment
> backed into it (or, makes this configurable).
Yes, that is true. On the other hand that lacks some
user-friendliness,
because ideally, you don't want to worry about expansion
services,
mostly when it comes to some mostly standard IO. The ideal
case is that
you either do not basically know that you use external
transform (which
is probably the case when you can use docker), or you are
able to
overcome the problem within the SDK (Python) by passing some
argument to
the input transform.
Arguments passed to the pipeline level apply to the whole
pipeline (not just one transform). So if you pass in a default
environment (and configs) at pipeline level, that would mean the
default environment and configs used by the pipeline (so Python
SDK in this case) not a specific transform.
I believe we have made usage of external transforms used-friendly
for the general case. But we had to make some assumptions. For
example we assumed,
* user will be using the default environment of the expansion
service (Docker in this case)
* User will be using the pre-specified dependency only
(sdks:java:io:expansion-service:shadowJar for Kafka)
* User will be in an environment where the jar can be downloaded.
I would consider any use-case where these basic assumptions
cannot be met as an advanced use-case. The solution in such a
case would be to start a custom expansion service and pass the
address of it as a parameter to the transform [1]. I'm fine with
extending the capabilities of Java expansion service by adding
more parameters (for example, for overriding the environment, for
specifying dependencies, for providing pipeline options).
Thanks,
Cham
[1]
https://github.com/apache/beam/blob/b86fcf94af26a240777f30f8193a314cb7ffc87e/sdks/python/apache_beam/io/kafka.py#L133
<https://github.com/apache/beam/blob/b86fcf94af26a240777f30f8193a314cb7ffc87e/sdks/python/apache_beam/io/kafka.py#L133>
>
> Is option (1) updating the default expansion service such
that one can
> override default environment properties on the command
line? (You
> would still have to start it up manually to use it.)
Yes and no. :) Updating ExpansionService so that you can
specify default
environment on command like makes this accessible to
JavaJarExpansionService, and that makes it possible to add
(optional)
argument to Python Kafka IO, that would delegate this to the
(automatically) started expansion service. It is important to
note that
both ReadFromKafka and WriteToKafka have expansion that
involves only
single external (Java) SDK. That simplifies things.
>
> Maybe it would help to make things more concrete. Suppose I
have a Go
> pipeline that uses a library which invokes a Python
external transform
> to do ML (say, via TFX), and two Java IOs (which happen to have
> mutually exclusive dependencies). The ML transform itself
uses Java to
> invoke some SQL.
>
> The way things work currently is each external transform
will have an
> associated fully specified environment and a runner can use
docker to
> start up the required workers at the expected time.
>
> Now, suppose one doesn't have docker on the workers. One
wants to run this with
>
> ./my_pipeline --someFlag=someValue
--someOtherFlag=someOtherValue ...
>
> such that docker is no longer needed. What someFlags would
we need,
> and what would their values be? (And how to make this
feasible to
> implement.)
>
> Are there meaningful intermediate points that extend to a
general
> solution (or at least aren't hostile to it)?
I believe that in the option 2) the best way would to use
each SDK's URN
Then the arguments could be something like
"--expansionEnvironments={"apache:beam:go:2.33.0:latest"={"env"="DOCKER",
config="<image>"},
"apache:beam:python:2.33.0:latest"={env="PROCESS",
config={...}}". Yes, it would require a lot of "syntactic
sugar" to
configure that. :) (sorry if I don't have URNs for SDKs 100%
correct)
>
>
> I still think in the long run having runners understand
environments,
> and saying "oh, whenever I see
'apache:beam:java:2.33.0:latest' I'll
> swap that out for 'path/to/my/java -cp ...' is the right
way to go
> long-term. (I would put this in runners, not SDKs, though a
common
> runners library could be used.)
Yes, I also agree, that expansion service should be
runner-dependent (or
at least runner-aware), as that brings optimizations. Runner
could
ignore settings from previous point when it can be *sure* it
can do so.
>
>
> On Tue, Jun 29, 2021 at 1:29 PM Jan Lukavský
<je...@seznam.cz <mailto:je...@seznam.cz>> wrote:
>> Thanks for pointing to that thread.
>>
>> 1) I'm - as well as Kyle - fine with the approach that we
use a
>> "preferred environment" for the expansion service. We only
need to pass
>> it via command line. Yes, the command line might be generally
>> SDK-dependent, and that makes it expansion dependent,
because whether or
>> not particular transform is "external" or not is
implementation detail.
>> That is the nasty part. The rest of my original question
is about, how
>> exactly to do that, because it seems to be tricky, due to
the fact, that
>> it is not possible to include runtime dependency on
DirectRunner (fails
>> many, many tests) and it is not possible to extract
PipelineOptions as a
>> Map either.
>>
>> 2) Regarding SDK injecting environment, I still think that
is the
>> correct way. The SDK (the driver code) own the execution
environment. It
>> should be able to define (or at least prioritize) runtime
environments
>> of all transforms. If we cannot know in advance, which
transform is
>> going to expand to how many nested (and possibly external)
transforms, I
>> think that the SDK could be fine with providing a Map(SDK ->
>> environment). That is: "Run Java using PROCESS", "Run
Python using
>> DOCKER", and so on. A default mapping might exist on the
expansion
>> service as well (which might be passed through command
line and that is
>> the point 1)). Yes, the Map approach is definitely not
universal,
>> because one can imagine that the SDK itself is not enough
for specifying
>> the environment, but seems that vast majority of cases
would fit into that.
>>
>> 3) The best might be for the SDK to provide a list of
supported
>> environments with additional metrics which the expansion
service might
>> choose from.
>>
>> These three approaches are all extensions to the current
state. Current
>> state has predefined environment without possibility to
change it.
>> Option 1) changes it to single configurable environment,
option 2) to N
>> environments based on SDK and option 3) to M environments
based on
>> SDK-dependent metrics (and/or capabilitites of particular
environment).
>> Seems like gradual extensions of the current state, so
maybe we can
>> focus on the first one, and maybe add other, when there is
a need?
>>
>> If this could be the first conclusion, then the next one
would be, what
>> should be the preferred way to implement it.
>>
>> WDYT?
>>
>> On 6/29/21 9:15 PM, Robert Bradshaw wrote:
>>> +1, thanks for digging up that thread.
>>>
>>> I am still of the same opinion that I wrote there. To
touch on some
>>> things brought up here, copying something like
>>> defaultEnvironmentConfig doesn't make sense from language
to language
>>> (e.g. the docker image name or CLI arguments for
subprocess mode just
>>> isn't going to work for all of Python, Java, and Go, and
clearly
>>> embedded type is only going to work for one.)
>>>
>>> In the short term, to change environment (or anything
else) about the
>>> "default" expansions service, the thing to do is build
and start your
>>> own expansion service that sets up the environment for
its transforms
>>> in a custom way.
>>>
>>> FYI, in Python, one can use --beam_services to use a
custom expansion
>>> service. E.g.
>>>
>>>
--beam_services='{":sdks:java:extensions:sql:expansion-service:shadowJar":
>>> "localhost:port"}'
>>>
>>> would override the default one when using SqlTransform.
>>>
>>> On Tue, Jun 29, 2021 at 11:47 AM Kyle Weaver
<kcwea...@google.com <mailto:kcwea...@google.com>> wrote:
>>>> For context, there was a previous thread which touched
on many of the same points:
https://lists.apache.org/thread.html/r6f6fc207ed62e1bf2a1d41deeeab554e35cd2af389ce38289a303cea%40%3Cdev.beam.apache.org%3E
<https://lists.apache.org/thread.html/r6f6fc207ed62e1bf2a1d41deeeab554e35cd2af389ce38289a303cea%40%3Cdev.beam.apache.org%3E>
>>>>
>>>> On Tue, Jun 29, 2021 at 11:16 AM Jan Lukavský
<je...@seznam.cz <mailto:je...@seznam.cz>> wrote:
>>>>> I would slightly disagree that this breaks the black
box nature of the expansion, the "how the transform expands"
remains unknown to the SDK requesting the expansion, the "how
the transform executes" - on the other hand - is something
that the SDK must cooperate on - it knows (or could or should
know) what is the environment that the pipeline is going to
be executed on looks like. That is why expansion service on
its own cannot correctly define the execution environment. It
could, if it would be bound to runner (and its environemnt) -
for instance FlinkRunnerExpansionService could probably
expand KafkaIO to something more 'native'. But that requires
knowledge of the target runner. If the expansion service is
not dedicated to a runner, the only place where it can be
defined, is the SDK - and therefore the expansion request.
>>>>>
>>>>>> Power users can always modify the output produced by
the expansion service as well.
>>>>> I'm not sure if I follow this, do you mean that power
users, who run the expansion service can modify the output?
Or is the output (protobuf) of the expansion service easily
transferable between different execution environments?- I had
the impression, that execution environments do not
necessarily have to have the same payloads associated with
them, and therefore it is impossible to 'postprocess' the
output of the expansion. Is that wrong assumption?
>>>>>
>>>>> On 6/29/21 7:55 PM, Luke Cwik wrote:
>>>>>
>>>>> This would "break" the black box where the expansion
service is supposed to hide the implementation internals from
the caller and pushes compatibility of these kinds of
environment overrides on to the expansion service and its
implementer.
>>>>>
>>>>> Power users can always modify the output produced by
the expansion service as well.
>>>>>
>>>>> On Tue, Jun 29, 2021 at 10:08 AM Jan Lukavský
<je...@seznam.cz <mailto:je...@seznam.cz>> wrote:
>>>>>> The argument for being able to accept (possibly
ordered list of) execution environments is in that this could
make a single instance of execution service reusable by
various clients with different requirements. Moreover, the
two approaches are probably orthogonal - users could specify
'defaultExecutionEnvironment' for the service which could be
used in case when there is no preference given by the client.
>>>>>>
>>>>>> On 6/29/21 7:03 PM, Luke Cwik wrote:
>>>>>>
>>>>>> I would be much more inclined for the user being able
to configure the expansion service for their needs instead of
changing the expansion service API.
>>>>>>
>>>>>> On Tue, Jun 29, 2021 at 9:42 AM Jan Lukavský
<je...@seznam.cz <mailto:je...@seznam.cz>> wrote:
>>>>>>> If I understand it correctly, there is currently no
place to set the
>>>>>>> defaultEnvironmentType - python's KafkaIO uses either
>>>>>>> 'expansion_service' given by the user (which might be
a host:port, or an
>>>>>>> object that has appropriate method), or calls
>>>>>>> 'default_io_expansion_service' - which in turn runs
ExpansionService
>>>>>>> using gradle. Either way, it ends up in
ExpansionService#main [1]. It
>>>>>>> could be possible to adapt ExpansionService and call
it locally -
>>>>>>> provided ExpansionService would provide a way to
extend it (using
>>>>>>> protected method createPipeline()) seems to be enough
- but that is not
>>>>>>> too much user-friendly. If we could specify the
defaultEnvironmentConfig
>>>>>>> when starting the ExpansionService, it would be
possible to add these
>>>>>>> parameters in the python SDK's KafkaIO, which would
mean users do not
>>>>>>> have to worry about the expansion service at all
(leaving aside that
>>>>>>> using too many ReafFromKafka or WriteToKafka
transforms would somewhat
>>>>>>> hurt performance during pipeline build, but that
applies to the pipeline
>>>>>>> build time only). I have created [2] to track that.
>>>>>>>
>>>>>>> Does that make sense, or is my analysis incorrect?
>>>>>>>
>>>>>>> Jan
>>>>>>>
>>>>>>> [1]
>>>>>>>
https://github.com/apache/beam/blob/22205ee1a84581e9206c5c61bad88a799779b4bc/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java#L511
<https://github.com/apache/beam/blob/22205ee1a84581e9206c5c61bad88a799779b4bc/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java#L511>
>>>>>>>
>>>>>>> [2] https://issues.apache.org/jira/browse/BEAM-12539
<https://issues.apache.org/jira/browse/BEAM-12539>
>>>>>>>
>>>>>>>
>>>>>>> On 6/29/21 6:24 PM, Alexey Romanenko wrote:
>>>>>>>> I’m sorry if I missed something but do you mean that
PortablePipelineOptions.setDefaultEnvironmentType(String)
doesn’t work for you? Or it’s only a specific case while
using portable KafkaIO?
>>>>>>>>
>>>>>>>>> On 29 Jun 2021, at 09:51, Jan Lukavský
<x666je...@gmail.com <mailto:x666je...@gmail.com>> wrote:
>>>>>>>>>
>>>>>>>>> Hi,
>>>>>>>>>
>>>>>>>>> I have come across an issue with cross-language
transforms. My setup is I have working environment type
PROCESS and I cannot use DOCKER. When I use Python's KafkaIO,
it unfortunately - by default - expands to docker
environment, which then fails due to missing 'docker'
command. I didn't find a solution without tackling the
expansion service, yet.
>>>>>>>>>
>>>>>>>>> I see several possible solutions to that:
>>>>>>>>>
>>>>>>>>> a) I would say, that the cleanest solution
would be to add preferred environment type to the expansion
request to the expansion service (probably along with
additional flags, probably --experiments?). This requires
deeper changes to the expansion RPC defintion, probably
serializing the PipelineOptions from the client environment
into the ExpansionRequest.
>>>>>>>>>
>>>>>>>>> b) Another option would be to allow specifying
some of the command-line arguments when starting the
expansion service, which currently accepts only port on
command line, see [1]. The straightforward 'fix' (see [2])
unfortunately does not work, because it requires DirectRunner
to be on the classpath, which then breaks other runners (see
[3]). It seems possible to copy hand selected options from
command line to the Pipeline, but that feels hackish. It
would require to either be able to construct the Pipeline
without a runner specified (which seems possible when calling
Pipeline.create(), but not when using PipelineOptions create
by parsing command-line arguments) or to be able to create a
Map<String, String> from PIpelineOptions and then the ability
to copy all options into the Pipeline's options.
>>>>>>>>>
>>>>>>>>> My proposal would be to create a hackish shortcut
and just copy the --defaultEnvironmentType,
--defaultEnvironmentConfig and --experiments into Pipeline's
options for now, and create an issue for a proper solution
(possible a)?).
>>>>>>>>>
>>>>>>>>> WDYT? Or did I miss a way to override the default
expansion?
>>>>>>>>>
>>>>>>>>> Thanks for comments,
>>>>>>>>>
>>>>>>>>> Jan
>>>>>>>>>
>>>>>>>>> [1]
https://github.com/apache/beam/blob/22205ee1a84581e9206c5c61bad88a799779b4bc/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java#L511
<https://github.com/apache/beam/blob/22205ee1a84581e9206c5c61bad88a799779b4bc/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java#L511>
>>>>>>>>>
>>>>>>>>> [2] https://github.com/apache/beam/pull/15082
<https://github.com/apache/beam/pull/15082>
>>>>>>>>>
>>>>>>>>> [3]
https://ci-beam.apache.org/job/beam_PreCommit_Java_Commit/18169/
<https://ci-beam.apache.org/job/beam_PreCommit_Java_Commit/18169/>
>>>>>>>>>
