On Wed, Jun 30, 2021 at 8:48 AM Jan Lukavský <je...@seznam.cz
<mailto:je...@seznam.cz>> wrote:
> Not sure why we need the hacks with NoOpRunner
As noted earlier (and that was why I started this thread in
the first
place :)), adding :runners:direct-java as runtime dependency
of the
expansion service causes something like 200 tests in
pre-commit to fail.
Looks like there is some kind of conflict among Flink and
Direct runner.
I didn't dig too deep into that, though.
> You could use the Python utilities in your script to
start/stop it
manually.
Yes, that is possible. I'll probably follow that path.
> This is where the runner's ability to customize
environments would
come in handy--e.g. a Java runner could decide to swap out
the Java
docker environment for EMBEDDED or LOOPBACK (and a
Python-based runner
could do the same for the Python docker env).
That would be just perfect, as that would make it possible
to finally
unify 'classical' and 'portable' runners. But that is a
whole different
story. :)
Jan
On 6/30/21 5:35 PM, Robert Bradshaw wrote:
> On Wed, Jun 30, 2021 at 7:41 AM Jan Lukavský
<je...@seznam.cz <mailto:je...@seznam.cz>> wrote:
>>> java -jar beam-sdks-java-io-expansion-service-2.30.0.jar
<port>
>> This does not accept any other parameters than the port.
That is the first part of this thread - the intent was to
enable this to accept additional arguments, but there are
(still waiting to be addressed unresolved) issues. There
currently even seems to be no other way to adapt
ExpansionService than to copy&paste the code and modify it,
because it simply is not extensible. What would be enough is
wrapping Pipeline.create() [1] call to a protected method,
or add (protected) constructor that would accept
PipelineOptions (probably better in this regard). That would
make it more easy for users to create customized
ExpansionService and it would (sort of) help solving
described issues.
> Yes, let's make it easy to extend/customize/start up a custom
> ExpansionService, including adding optional command line
arguments to
> the "default" one. Not sure why we need the hacks with
NoOpRunner
> (IMHO, the direct runner should just be part of the SDK,
but that's
> not where we live now).
>
>> But even if we do that, we still need to deal with the
expansion service on two places:
>>
>> a) run it (and stop it)
>>
>> b) specify it in the
>>
>> Using the default expansion service is much, much easier,
it is started and stopped automatically for the user.
Morever, the JavaJarExpansionService actually even presumes
that there can be additional arguments passed to the service
([2]), the ExpansionService only does not accept them (and
kafka IO does not expose that - that could be worked-around
by users by manually creating the JavaJarExpansionService
from own jar, yes). I would find it natural to add the
command-line parsing (somehow!) to the ExpansionService
itself, so that it doesn't need end-user modifications and
then to figure out how to most easily expose there
command-line arguments to end-users.
> You could use the Python utilities in your script to
start/stop it manually.
>
>> Yes, I verified that Flink can use Python Kafka IO over
PROCESS environment with some hacking of the
ExpansionService as shown in one of the linked PRs (though
there is probably still some bugs regarding SDF - [3]).
Adding --experiments seems have the same issues, need expose
that to the CLI of ExpansionService. And I'm not sure if
this [4] is not in conflict with
--experiments=use_deprecated_read. That is something I still
need to investigate.
>>
>> LOOPBACK is currently not supported by Flink. That is
nice-to-have feature.
> Local Flink does support LOOPBACK mode. If you just want
to run
> locally, just specifying "FlinkRunner" is enough. It's
distributed
> Flink that does not. It seems a lot of complexities are
due to trying
> to using minikube, which acts like it's distributed, but
trying to
> make it as easy as if it were all local (and the docker
deficiencies
> as well, which would make it just work...) Which is a
worthy goal.
>
> This is where the runner's ability to customize
environments would
> come in handy--e.g. a Java runner could decide to swap out
the Java
> docker environment for EMBEDDED or LOOPBACK (and a
Python-based runner
> could do the same for the Python docker env).
>
>> [1]
https://github.com/apache/beam/blob/f9a4bfcb027f2e3a8e32578adf49981aeef3586a/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java#L394
<https://github.com/apache/beam/blob/f9a4bfcb027f2e3a8e32578adf49981aeef3586a/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java#L394>
>>
>> [2]
https://github.com/apache/beam/blob/f9a4bfcb027f2e3a8e32578adf49981aeef3586a/sdks/python/apache_beam/transforms/external.py#L481
<https://github.com/apache/beam/blob/f9a4bfcb027f2e3a8e32578adf49981aeef3586a/sdks/python/apache_beam/transforms/external.py#L481>
>>
>> [3] https://issues.apache.org/jira/browse/BEAM-11998
<https://issues.apache.org/jira/browse/BEAM-11998>
>>
>> [4]
https://github.com/apache/beam/blob/b86fcf94af26a240777f30f8193a314cb7ffc87e/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java#L398
<https://github.com/apache/beam/blob/b86fcf94af26a240777f30f8193a314cb7ffc87e/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java#L398>
>>
>> On 6/30/21 3:57 PM, Chamikara Jayalath wrote:
>>
>>
>>
>> On Wed, Jun 30, 2021 at 6:54 AM Chamikara Jayalath
<chamik...@google.com <mailto:chamik...@google.com>> wrote:
>>>
>>>
>>> On Wed, Jun 30, 2021 at 1:20 AM Jan Lukavský
<je...@seznam.cz <mailto:je...@seznam.cz>> wrote:
>>>> On 6/30/21 1:16 AM, Robert Bradshaw wrote:
>>>>> <rant>Why doesn't docker in docker just work, rather
than having to do
>>>>> ugly hacks when composing two technologies that both
rely on
>>>>> docker...</rant>
>>>>>
>>>>> Presumably you're setting up a node for Kafka and
Flink; why not set
>>>>> one up for the expansion service as well? The UX of
>>>>>
>>>>>
ReadFromKafka(default_expansion_service_args={"defaultEnvironmentType":
>>>>> "PROCESS", "defaultEnvironmentConfig": ""{\"os\":
\"linux\", \"arch\":
>>>>> \"amd64\", \"command\": \"/path/to/launcher/boot
>>>>> cp=/some/other/long/path\" ...}")"})
>>>>>
>>>>> isn't that great either. Rather than pass arbitrary
arguments to a
>>>>> default expansion service, I still think once you get
to this level
>>>>> it's better to just start your own expansion service.
>>>> Sure, that is possible (seems to me, that it would
still require some
>>>> changes to ExpansionService to be extendable, but yes,
kind of tiny
>>>> changes). The problem is not with Flink or Kafka -
those are
>>>> technologies you are actually expecting to set up,
because you want to
>>>> use them. The problem is what everything else you must
set up for making
>>>> something that seems as easy as "read a few messages
from kafka in beam
>>>> python" to work. You must have:
>>>>
>>>> a) Python SDK harness (OK, that is something that
should be probably
>>>> expected) - there are few problems with it, namely it
is somewhat
>>>> hardcoded that it must run in the same pod as Flink's
taskmanager to be
>>>> able to use EXTERNAL environment, but ok, let's go on
>>>>
>>>> b) Java SDK harness, at least installed in docker
image of taskmanager
>>>> (to be usable via PROCESS environment) - OK, that
starts to be weird,
>>>> taskmanager is java, right? Something like LOOPBACK
would be cool there,
>>>> but never mind. You create custom docker image for your
Flink JM and TM
>>>> and continue.
>>>>
>>>> c) Implement (extend) and deploy own expansion
service - ouch, that
>>>> starts to hurt, that is even going to be a pod that is
running even
>>>> though there is nothing using it (yes, can be scaled down).
>>>>
>>>> The complexity of a simple task starts to be somewhat
extraordinary. And
>>>> most of the users will not be willing to follow this
path, I'm afraid.
>>>> People generally don't like to set up complex
environment for something
>>>> that looks it should "just work". There is non-trivial
work necessary
>>>> to make all of this working, mostly when you are
starting to evaluate
>>>> Beam and don't have much experience with it.
>>>
>>> I don't think we should expect end-users to implement or
extend the expansion service. Everything should be already
implemented and maybe we can even provide a script to easily
startup a local Java expansion service with additional
parameters.
>>>
>>> Today, to start a Java expansion service for Kafka users
have to do the following.
>>>
>>> * Download expansion service jar released with Beam for
Kafka. For example [1]
>>>
>>> * Run following command:
>>> java -jar beam-sdks-java-io-expansion-service-2.30.0.jar
<port>
>>>
>>> * To use this they just have to provide
"localhost:<port>" to [2].
>>>
>>> This is a few extra steps but mostly a one time setup
for the user and nothing to do with portability or other
complexities of Beam.
>>>
>>> I'm all for simplifying the user-experience, but adding
changes to the transform API that might have to be
deprecated later sounds like a bad idea. I'd much rather
provide additional scripts/documentation/examples to
simplify such use-cases. I think that will be adequate for
most users.
>>>
>>> BTW, slightly orthogonal, I don't think multi-language
would work in LOOPBACK mode today without additional changes
to portable runners (at least I've never tested this). Did
you confirm that this works ?
>>
>> Or PROCESS mode.
>>
>>>
>>> Thanks,
>>> Cham
>>>
>>> [1]
https://repo1.maven.org/maven2/org/apache/beam/beam-sdks-java-io-expansion-service/2.30.0/beam-sdks-java-io-expansion-service-2.30.0.jar
<https://repo1.maven.org/maven2/org/apache/beam/beam-sdks-java-io-expansion-service/2.30.0/beam-sdks-java-io-expansion-service-2.30.0.jar>
>>> [2]
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>
>>>
>>>
>>>>
>>>> We can get rid of b) (implement LOOPBACK in Flink) and
c) (enable Python
>>>> SDK Kafka IO to spawn expansion service with the
LOOPBACK environment
>>>> when submitting to Flink). That is why I still think
that this
>>>> simplification matters a lot.
>>>>
>>>>> On Tue, Jun 29, 2021 at 3:33 PM Jan Lukavský
<je...@seznam.cz <mailto:je...@seznam.cz>> wrote:
>>>>>> I believe we could change that more or less the same
as we can deprecate / stop supporting any other parameter of
any method. If python starts to support natively Kafka IO,
then we can simply log warning / raise exception (one after
the other). That seems like natural development.
>>>>>>
>>>>>> Maybe I should have described the case - I'm trying
to setup a "simple" use-case for users that want to try
Python SDK to read using Flink from Kafka using Minikube
(both Kafka and Flink are running inside Minikube). There
are tons of problems to use docker from within Minkube and I
would not say that is the "simple" way we would like to
present to users. Setting up own expansion service is
possibility - but that also lacks the UX approach. I pretty
much think that understanding portability on it's own is
already a burden we put on users (yes, we do that for a
reason, but everything else should be as simple as possible).
>>>>>>
>>>>>> On 6/30/21 12:16 AM, Chamikara Jayalath wrote:
>>>>>>
>>>>>> So I think one downside to this PR is that we assume
that the default expansion service used by the transform
(Kafka in this case) will not change. Currently it's fully
opaque. In the default case we just promise that the
transform will work (if conditions I mentioned above are
met). Nothing else.
>>>>>> If we add a "param default_expansion_service_args",
we leak the nature of the default expansion service to the
API and it will be hard to change it in the future.
>>>>>>
>>>>>> Thanks,
>>>>>> Cham
>>>>>>
>>>>>> On Tue, Jun 29, 2021 at 3:07 PM Jan Lukavský
<je...@seznam.cz <mailto:je...@seznam.cz>> wrote:
>>>>>>> I would absolutely understand this, if it would be
mostly impossible or at least really hard to get the user
friendly behavior. But we are mostly there in this case.
When we can actually quite simply pass the supported
environment via parameter, I think we should go for it.
>>>>>>>
>>>>>>> I have created a sketch (I verified that when the
ExpansionService is patched 'enough' this works) in [1].
This is only a sketch, because we first must know how to
support the default execution environment in ExpansionService.
>>>>>>>
>>>>>>> [1] https://github.com/apache/beam/pull/15099/files
<https://github.com/apache/beam/pull/15099/files>
>>>>>>>
>>>>>>> On 6/29/21 11:51 PM, Chamikara Jayalath wrote:
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> 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/>
>>>>>>>>>>>>>>>>>
