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/>
>>>>>>>>>>>>>
