Stdin/stdout glue for external processes

[Dave Revell <da...@lytics.io>]

Hi all,

We're considering using Samza for our high-throughput stream processing
workload, but we don't want to rewrite all of our existing Go code. We're
considering writing something analogous to Hadoop Streaming, where the
Samza consumer would start an external process and communicate with it by
passing protobufs via stdin/stdout. We like Samza's fault tolerance, state
management, and load balancing features and don't want to rewrite them.

This possibility is mentioned in the documentation (
http://samza.incubator.apache.org/learn/documentation/0.7.0/comparisons/storm.html,
search for "stdin") as something that might exist some day. My
questions
are:

1. Is anyone working on this, or planning to? I couldn't find any related
JIRAs.
2. Any advice for implementing this? Are there any challenges that might
not be obvious?
3. Should we try to merge this upstream?

Thanks a bunch,
Dave

Re: Stdin/stdout glue for external processes

[Travis Brady <tr...@gmail.com>]

I initially requested SAMZA-18 some months ago and am still interested.
 Out of curiosity, what would such an interface look like?  Is this
something the core team could sketch out?

Travis




On Mon, Mar 10, 2014 at 2:26 PM, Jakob Homan <jg...@gmail.com> wrote:

> Hey Dave-
>    Thanks for taking a look at Samza.  No one in the community is currently
> working on this at the moment, to our knowledge.  SAMZA-18 (
> https://issues.apache.org/jira/browse/SAMZA-18) has the beginnings of a
> discussion about creating a single C library to help provide multilanguage
> support in Samza (which I believe would be accessible to Go as well).
> There's currently no JIRA for Hadoop-style streaming, but one could
> certainly be created and it would be something we'd be interested in.
> Thanks,
> Jakob
>
>
>
> On Mon, Mar 10, 2014 at 12:19 PM, Dave Revell <da...@lytics.io> wrote:
>
> > Hi all,
> >
> > We're considering using Samza for our high-throughput stream processing
> > workload, but we don't want to rewrite all of our existing Go code. We're
> > considering writing something analogous to Hadoop Streaming, where the
> > Samza consumer would start an external process and communicate with it by
> > passing protobufs via stdin/stdout. We like Samza's fault tolerance,
> state
> > management, and load balancing features and don't want to rewrite them.
> >
> > This possibility is mentioned in the documentation (
> >
> >
> http://samza.incubator.apache.org/learn/documentation/0.7.0/comparisons/storm.html
> > ,
> > search for "stdin") as something that might exist some day. My
> > questions
> > are:
> >
> > 1. Is anyone working on this, or planning to? I couldn't find any related
> > JIRAs.
> > 2. Any advice for implementing this? Are there any challenges that might
> > not be obvious?
> > 3. Should we try to merge this upstream?
> >
> > Thanks a bunch,
> > Dave
> >
>

Re: Stdin/stdout glue for external processes

[Dave Revell <da...@lytics.io>]

OK, I'll plan to do a wiki page, as soon as I'm up to speed on Samza enough
to do a decent job :)

-Dave


On Tue, Mar 11, 2014 at 10:02 AM, Jay Kreps <ja...@gmail.com> wrote:

> Cool! A prototype would be good, but I think for this kind of thing where
> potentially many client langs would integrate it would make sense to do a
> quick wiki page on how the protocol might work and work that out up-front
> or in parallel...
>
> -Jay
>
>
> On Tue, Mar 11, 2014 at 9:54 AM, Dave Revell <da...@lytics.io> wrote:
>
> > OP here, thanks to everyone for your thoughts. Replies inline:
> >
> > In reply to Jakob, thanks for the pointer to SAMZA-18, that's good to
> know.
> > I probably won't hold my breath waiting for it though :)
> >
> > In reply to Martin:
> >
> > > Were you envisaging that such API calls could be made over a protocol
> > > that uses stdout and stdin as its transport? I'm sure it can be done,
> but
> > > the protocol would not be totally straightforward, as it would have to
> do
> > > things like match a request from the child process (on stdout) with a
> > > response to the API call (on stdin).
> >
> > Samza currently has the nice property of having one message in flight at
> a
> > time per partition. I think we'd want to preserve that and avoid adding
> any
> > pipelining by having multiple messages in flight. So that means we could
> > use a synchronous protocol over stdin/stdout that doesn't need to match
> > requests to responses. Please correct me if I'm wrong on this, I'm still
> > very new to Samza.
> >
> > In reply to Jay, I take your point about Hadoop Streaming. Adding
> > lifecycle/control messages to the protocol seems beneficial. We could
> also
> > allow the external process to access the KV store in the JVM via the
> > stdin/stdout protocol. We'd also want to use an encoding scheme that
> > doesn't have problems with separators (Hadoop Streaming has trouble with
> > data that contains internal tabs and newlines). I think I'd prefer
> > protobufs over JSON for performance reasons, but it could be
> configurable.
> >
> > So it seems like this is doable. There are some design questions but
> maybe
> > a proof of concept would be a good next step. We'll be in touch if/when
> > that happens. If anyone else wants to try it, we'd welcome that also :)
> >
> > Cheers,
> > Dave
> >
> >
> >
> >
> > On Tue, Mar 11, 2014 at 8:58 AM, Jay Kreps <ja...@gmail.com> wrote:
> >
> > > Yes, I think all I am saying is that stdin/stdout aren't so bad. The
> > > mistake made by Hadoop streaming I think was to not specify a more
> > detailed
> > > extensible control protocol for the client process. (or at least that
> > > seemed to be true a long-ass time ago when I last used Hadoop
> streaming).
> > > You would presumably model commands like commit as outputs and config
> and
> > > such as input and you would need so predetermined data format to read
> and
> > > write the topic/partition/key/value pairs. Sounds like Storm did a
> better
> > > job here.
> > >
> > > As you say this protocol needn't go over stdin/out but it is fairly
> cheap
> > > and I think unix domain sockets are not well supported in Java.
> > >
> > > As you say you definitely don't get the full power out of the box. For
> > > example the key-value store would have to be in the child process and
> all
> > > we could provide would be the backing changelog stream.
> > >
> > > Basically I am saying the maintenance burden of doing this seems
> > low--it's
> > > just a simple samza job that manages a native subprocess and feeds it
> > > formatted input--so there is no harm in pursuing both approaches...
> > >
> > > -Jay
> > >
> > >
> > > On Mon, Mar 10, 2014 at 6:10 PM, Martin Kleppmann
> > > <mk...@linkedin.com>wrote:
> > >
> > > > If we want to make non-JVM languages first-class citizens in Samza
> > (which
> > > > I think we should), they will need access to all the Samza APIs,
> > > including
> > > > reading and writing the key-value store and the like.
> > > >
> > > > Were you envisaging that such API calls could be made over a protocol
> > > that
> > > > uses stdout and stdin as its transport? I'm sure it can be done, but
> > the
> > > > protocol would not be totally straightforward, as it would have to do
> > > > things like match a request from the child process (on stdout) with a
> > > > response to the API call (on stdin).
> > > >
> > > > Then there is a separate question of what transport the non-JVM
> process
> > > > uses to talk to the Samza container. stdout/stdin streams is an
> option,
> > > or
> > > > it could use Unix sockets, or TCP. Whatever method of transport is
> > used,
> > > a
> > > > protocol will need to encode the Samza API calls in raw bytes.
> > > >
> > > > For reference, Storm uses a JSON-based protocol over stdin/stdout:
> > > > https://github.com/nathanmarz/storm/wiki/Multilang-protocol --
> > however,
> > > > it doesn't support all Storm features (
> > > > https://issues.apache.org/jira/browse/STORM-151).
> > > >
> > > > Martin
> > > >
> > > > On 10 Mar 2014, at 23:21, Jay Kreps <ja...@gmail.com> wrote:
> > > > > FWIW, I actually think the Hadoop streaming approach has some
> > benefits.
> > > > It
> > > > > is less efficient then writing and embedding a C library but also
> > much
> > > > much
> > > > > easier to implement and with less duplicate logic. I think we
> should
> > be
> > > > > open to both of these--the streaming approach is so easy, it seems
> to
> > > me
> > > > > like there is not a huge downside to having that available.
> > > > >
> > > > > I think the mistake that Hadoop streaming might have made was
> > > > > over-simplifying the interaction with the client process. You
> > probably
> > > > need
> > > > > a richer protocol than just the data (though I haven't thought this
> > > > > through).
> > > > >
> > > > > -Jay
> > > > >
> > > > >
> > > > > On Mon, Mar 10, 2014 at 12:26 PM, Jakob Homan <jg...@gmail.com>
> > > wrote:
> > > > >
> > > > >> Hey Dave-
> > > > >>   Thanks for taking a look at Samza.  No one in the community is
> > > > currently
> > > > >> working on this at the moment, to our knowledge.  SAMZA-18 (
> > > > >> https://issues.apache.org/jira/browse/SAMZA-18) has the
> beginnings
> > > of a
> > > > >> discussion about creating a single C library to help provide
> > > > multilanguage
> > > > >> support in Samza (which I believe would be accessible to Go as
> > well).
> > > > >> There's currently no JIRA for Hadoop-style streaming, but one
> could
> > > > >> certainly be created and it would be something we'd be interested
> > in.
> > > > >> Thanks,
> > > > >> Jakob
> > > > >>
> > > > >>
> > > > >>
> > > > >> On Mon, Mar 10, 2014 at 12:19 PM, Dave Revell <da...@lytics.io>
> > wrote:
> > > > >>
> > > > >>> Hi all,
> > > > >>>
> > > > >>> We're considering using Samza for our high-throughput stream
> > > processing
> > > > >>> workload, but we don't want to rewrite all of our existing Go
> code.
> > > > We're
> > > > >>> considering writing something analogous to Hadoop Streaming,
> where
> > > the
> > > > >>> Samza consumer would start an external process and communicate
> with
> > > it
> > > > by
> > > > >>> passing protobufs via stdin/stdout. We like Samza's fault
> > tolerance,
> > > > >> state
> > > > >>> management, and load balancing features and don't want to rewrite
> > > them.
> > > > >>>
> > > > >>> This possibility is mentioned in the documentation (
> > > > >>>
> > > > >>>
> > > > >>
> > > >
> > >
> >
> http://samza.incubator.apache.org/learn/documentation/0.7.0/comparisons/storm.html
> > > > >>> ,
> > > > >>> search for "stdin") as something that might exist some day. My
> > > > >>> questions
> > > > >>> are:
> > > > >>>
> > > > >>> 1. Is anyone working on this, or planning to? I couldn't find any
> > > > related
> > > > >>> JIRAs.
> > > > >>> 2. Any advice for implementing this? Are there any challenges
> that
> > > > might
> > > > >>> not be obvious?
> > > > >>> 3. Should we try to merge this upstream?
> > > > >>>
> > > > >>> Thanks a bunch,
> > > > >>> Dave
> > > > >>>
> > > > >>
> > > >
> > > >
> > >
> >
>

Re: Stdin/stdout glue for external processes

[Jay Kreps <ja...@gmail.com>]

Cool! A prototype would be good, but I think for this kind of thing where
potentially many client langs would integrate it would make sense to do a
quick wiki page on how the protocol might work and work that out up-front
or in parallel...

-Jay


On Tue, Mar 11, 2014 at 9:54 AM, Dave Revell <da...@lytics.io> wrote:

> OP here, thanks to everyone for your thoughts. Replies inline:
>
> In reply to Jakob, thanks for the pointer to SAMZA-18, that's good to know.
> I probably won't hold my breath waiting for it though :)
>
> In reply to Martin:
>
> > Were you envisaging that such API calls could be made over a protocol
> > that uses stdout and stdin as its transport? I'm sure it can be done, but
> > the protocol would not be totally straightforward, as it would have to do
> > things like match a request from the child process (on stdout) with a
> > response to the API call (on stdin).
>
> Samza currently has the nice property of having one message in flight at a
> time per partition. I think we'd want to preserve that and avoid adding any
> pipelining by having multiple messages in flight. So that means we could
> use a synchronous protocol over stdin/stdout that doesn't need to match
> requests to responses. Please correct me if I'm wrong on this, I'm still
> very new to Samza.
>
> In reply to Jay, I take your point about Hadoop Streaming. Adding
> lifecycle/control messages to the protocol seems beneficial. We could also
> allow the external process to access the KV store in the JVM via the
> stdin/stdout protocol. We'd also want to use an encoding scheme that
> doesn't have problems with separators (Hadoop Streaming has trouble with
> data that contains internal tabs and newlines). I think I'd prefer
> protobufs over JSON for performance reasons, but it could be configurable.
>
> So it seems like this is doable. There are some design questions but maybe
> a proof of concept would be a good next step. We'll be in touch if/when
> that happens. If anyone else wants to try it, we'd welcome that also :)
>
> Cheers,
> Dave
>
>
>
>
> On Tue, Mar 11, 2014 at 8:58 AM, Jay Kreps <ja...@gmail.com> wrote:
>
> > Yes, I think all I am saying is that stdin/stdout aren't so bad. The
> > mistake made by Hadoop streaming I think was to not specify a more
> detailed
> > extensible control protocol for the client process. (or at least that
> > seemed to be true a long-ass time ago when I last used Hadoop streaming).
> > You would presumably model commands like commit as outputs and config and
> > such as input and you would need so predetermined data format to read and
> > write the topic/partition/key/value pairs. Sounds like Storm did a better
> > job here.
> >
> > As you say this protocol needn't go over stdin/out but it is fairly cheap
> > and I think unix domain sockets are not well supported in Java.
> >
> > As you say you definitely don't get the full power out of the box. For
> > example the key-value store would have to be in the child process and all
> > we could provide would be the backing changelog stream.
> >
> > Basically I am saying the maintenance burden of doing this seems
> low--it's
> > just a simple samza job that manages a native subprocess and feeds it
> > formatted input--so there is no harm in pursuing both approaches...
> >
> > -Jay
> >
> >
> > On Mon, Mar 10, 2014 at 6:10 PM, Martin Kleppmann
> > <mk...@linkedin.com>wrote:
> >
> > > If we want to make non-JVM languages first-class citizens in Samza
> (which
> > > I think we should), they will need access to all the Samza APIs,
> > including
> > > reading and writing the key-value store and the like.
> > >
> > > Were you envisaging that such API calls could be made over a protocol
> > that
> > > uses stdout and stdin as its transport? I'm sure it can be done, but
> the
> > > protocol would not be totally straightforward, as it would have to do
> > > things like match a request from the child process (on stdout) with a
> > > response to the API call (on stdin).
> > >
> > > Then there is a separate question of what transport the non-JVM process
> > > uses to talk to the Samza container. stdout/stdin streams is an option,
> > or
> > > it could use Unix sockets, or TCP. Whatever method of transport is
> used,
> > a
> > > protocol will need to encode the Samza API calls in raw bytes.
> > >
> > > For reference, Storm uses a JSON-based protocol over stdin/stdout:
> > > https://github.com/nathanmarz/storm/wiki/Multilang-protocol --
> however,
> > > it doesn't support all Storm features (
> > > https://issues.apache.org/jira/browse/STORM-151).
> > >
> > > Martin
> > >
> > > On 10 Mar 2014, at 23:21, Jay Kreps <ja...@gmail.com> wrote:
> > > > FWIW, I actually think the Hadoop streaming approach has some
> benefits.
> > > It
> > > > is less efficient then writing and embedding a C library but also
> much
> > > much
> > > > easier to implement and with less duplicate logic. I think we should
> be
> > > > open to both of these--the streaming approach is so easy, it seems to
> > me
> > > > like there is not a huge downside to having that available.
> > > >
> > > > I think the mistake that Hadoop streaming might have made was
> > > > over-simplifying the interaction with the client process. You
> probably
> > > need
> > > > a richer protocol than just the data (though I haven't thought this
> > > > through).
> > > >
> > > > -Jay
> > > >
> > > >
> > > > On Mon, Mar 10, 2014 at 12:26 PM, Jakob Homan <jg...@gmail.com>
> > wrote:
> > > >
> > > >> Hey Dave-
> > > >>   Thanks for taking a look at Samza.  No one in the community is
> > > currently
> > > >> working on this at the moment, to our knowledge.  SAMZA-18 (
> > > >> https://issues.apache.org/jira/browse/SAMZA-18) has the beginnings
> > of a
> > > >> discussion about creating a single C library to help provide
> > > multilanguage
> > > >> support in Samza (which I believe would be accessible to Go as
> well).
> > > >> There's currently no JIRA for Hadoop-style streaming, but one could
> > > >> certainly be created and it would be something we'd be interested
> in.
> > > >> Thanks,
> > > >> Jakob
> > > >>
> > > >>
> > > >>
> > > >> On Mon, Mar 10, 2014 at 12:19 PM, Dave Revell <da...@lytics.io>
> wrote:
> > > >>
> > > >>> Hi all,
> > > >>>
> > > >>> We're considering using Samza for our high-throughput stream
> > processing
> > > >>> workload, but we don't want to rewrite all of our existing Go code.
> > > We're
> > > >>> considering writing something analogous to Hadoop Streaming, where
> > the
> > > >>> Samza consumer would start an external process and communicate with
> > it
> > > by
> > > >>> passing protobufs via stdin/stdout. We like Samza's fault
> tolerance,
> > > >> state
> > > >>> management, and load balancing features and don't want to rewrite
> > them.
> > > >>>
> > > >>> This possibility is mentioned in the documentation (
> > > >>>
> > > >>>
> > > >>
> > >
> >
> http://samza.incubator.apache.org/learn/documentation/0.7.0/comparisons/storm.html
> > > >>> ,
> > > >>> search for "stdin") as something that might exist some day. My
> > > >>> questions
> > > >>> are:
> > > >>>
> > > >>> 1. Is anyone working on this, or planning to? I couldn't find any
> > > related
> > > >>> JIRAs.
> > > >>> 2. Any advice for implementing this? Are there any challenges that
> > > might
> > > >>> not be obvious?
> > > >>> 3. Should we try to merge this upstream?
> > > >>>
> > > >>> Thanks a bunch,
> > > >>> Dave
> > > >>>
> > > >>
> > >
> > >
> >
>

Re: Stdin/stdout glue for external processes

[Dave Revell <da...@lytics.io>]

Hi Chris,

Thanks for opening the ticket. I'll definitely create a wiki page about the
proposed design and protocol. Currently I'm still getting my feet wet with
Samza and working to understand its interfaces and capabilities. Once I
feel like I have a basic grasp of how all the pieces should fit together
I'll put up a wiki page and send a message to this thread.

Thanks,
Dave


On Thu, Mar 13, 2014 at 1:19 PM, Chris Riccomini <cr...@linkedin.com>wrote:

> Hey Guys,
>
> Sorry I've been silent on this. I think we should pursue this approach as
> a nice easy low friction way to support multiple languages. I've opened up
> a ticket:
>
>   https://issues.apache.org/jira/browse/SAMZA-184
>
> Dave, if you're still up for it, it'd be really awesome to get a Wiki
> proposal from you. Since you have an actual use case, it'll be useful to
> have you provide direction. I'm happy to advise and answer any questions
> you have. I'm going to post my initial thoughts as a follow up comment on
> SAMZA-184.
>
> Cheers,
> Chris
>
> On 3/12/14 9:41 AM, "Dave Revell" <da...@lytics.io> wrote:
>
> >Thanks a lot for the detailed reply. Responses inline:
> >
> >Agree -- it's easiest if there's only one message at a time being sent to
> >> the child process. Though we should benchmark that to make sure that
> >> performance is still good.
> >>
> >
> >Sounds good to me. The latency of the system calls + marshaling might turn
> >out to be significant.
> >
> >
> >> With one message at a time, a task always has to write something to
> >>stdout
> >> for every message it consumes, even if it doesn't want to emit an output
> >> for a particular input message -- otherwise Samza wouldn't know when to
> >> send the next message to the task.
> >>
> >
> >That's true, I didn't think of that. It shouldn't be hard though.
> >
> >
> >> Another thing to think about: do we want one child process per task, or
> >> one per container? One per task is a simpler processing model (matches
> >>the
> >> Java API), but one per container perhaps makes more sense from a
> >>resource
> >> allocation point of view.
> >>
> >
> >I think one per task. If the external process is stateful, that state
> >should be limited to a single partition. But I don't understand Samza well
> >enough to say whether this is the overriding concern. Also, like you say,
> >the programming model is much nicer.
> >
> >
> >> Yeah, I think allowing access to the KV store via stdin/stdout protocol
> >> makes the most sense. For example, to make a "get" request to the store,
> >> the task could write to stdout:
> >>
> >> {"cmd": "kv_get", "store": "my-store", "key": "foo"}
> >>
> >> to which Samza would respond by sending to stdin:
> >>
> >> {"cmd": "kv_get_response", "store": "my-store", "key": "foo", "value":
> >> "bar"}
> >>
> >
> >Sounds good to me.
> >
> >Thanks again,
> >Dave
>
>

Re: Stdin/stdout glue for external processes

[Chris Riccomini <cr...@linkedin.com>]

Hey Guys,

Sorry I've been silent on this. I think we should pursue this approach as
a nice easy low friction way to support multiple languages. I've opened up
a ticket:

  https://issues.apache.org/jira/browse/SAMZA-184

Dave, if you're still up for it, it'd be really awesome to get a Wiki
proposal from you. Since you have an actual use case, it'll be useful to
have you provide direction. I'm happy to advise and answer any questions
you have. I'm going to post my initial thoughts as a follow up comment on
SAMZA-184.

Cheers,
Chris

On 3/12/14 9:41 AM, "Dave Revell" <da...@lytics.io> wrote:

>Thanks a lot for the detailed reply. Responses inline:
>
>Agree -- it's easiest if there's only one message at a time being sent to
>> the child process. Though we should benchmark that to make sure that
>> performance is still good.
>>
>
>Sounds good to me. The latency of the system calls + marshaling might turn
>out to be significant.
>
>
>> With one message at a time, a task always has to write something to
>>stdout
>> for every message it consumes, even if it doesn't want to emit an output
>> for a particular input message -- otherwise Samza wouldn't know when to
>> send the next message to the task.
>>
>
>That's true, I didn't think of that. It shouldn't be hard though.
>
>
>> Another thing to think about: do we want one child process per task, or
>> one per container? One per task is a simpler processing model (matches
>>the
>> Java API), but one per container perhaps makes more sense from a
>>resource
>> allocation point of view.
>>
>
>I think one per task. If the external process is stateful, that state
>should be limited to a single partition. But I don't understand Samza well
>enough to say whether this is the overriding concern. Also, like you say,
>the programming model is much nicer.
>
>
>> Yeah, I think allowing access to the KV store via stdin/stdout protocol
>> makes the most sense. For example, to make a "get" request to the store,
>> the task could write to stdout:
>>
>> {"cmd": "kv_get", "store": "my-store", "key": "foo"}
>>
>> to which Samza would respond by sending to stdin:
>>
>> {"cmd": "kv_get_response", "store": "my-store", "key": "foo", "value":
>> "bar"}
>>
>
>Sounds good to me.
>
>Thanks again,
>Dave

Re: Stdin/stdout glue for external processes

[Dave Revell <da...@lytics.io>]

Thanks a lot for the detailed reply. Responses inline:

Agree -- it's easiest if there's only one message at a time being sent to
> the child process. Though we should benchmark that to make sure that
> performance is still good.
>

Sounds good to me. The latency of the system calls + marshaling might turn
out to be significant.


> With one message at a time, a task always has to write something to stdout
> for every message it consumes, even if it doesn't want to emit an output
> for a particular input message -- otherwise Samza wouldn't know when to
> send the next message to the task.
>

That's true, I didn't think of that. It shouldn't be hard though.


> Another thing to think about: do we want one child process per task, or
> one per container? One per task is a simpler processing model (matches the
> Java API), but one per container perhaps makes more sense from a resource
> allocation point of view.
>

I think one per task. If the external process is stateful, that state
should be limited to a single partition. But I don't understand Samza well
enough to say whether this is the overriding concern. Also, like you say,
the programming model is much nicer.


> Yeah, I think allowing access to the KV store via stdin/stdout protocol
> makes the most sense. For example, to make a "get" request to the store,
> the task could write to stdout:
>
> {"cmd": "kv_get", "store": "my-store", "key": "foo"}
>
> to which Samza would respond by sending to stdin:
>
> {"cmd": "kv_get_response", "store": "my-store", "key": "foo", "value":
> "bar"}
>

Sounds good to me.

Thanks again,
Dave

Re: Stdin/stdout glue for external processes

[Martin Kleppmann <mk...@linkedin.com>]

Hi Dave,

On 11 Mar 2014, at 16:54, Dave Revell <da...@lytics.io> wrote:
> Samza currently has the nice property of having one message in flight at a
> time per partition. I think we'd want to preserve that and avoid adding any
> pipelining by having multiple messages in flight.

Agree -- it's easiest if there's only one message at a time being sent to the child process. Though we should benchmark that to make sure that performance is still good.

With one message at a time, a task always has to write something to stdout for every message it consumes, even if it doesn't want to emit an output for a particular input message -- otherwise Samza wouldn't know when to send the next message to the task.

Another thing to think about: do we want one child process per task, or one per container? One per task is a simpler processing model (matches the Java API), but one per container perhaps makes more sense from a resource allocation point of view.

> In reply to Jay, I take your point about Hadoop Streaming. Adding
> lifecycle/control messages to the protocol seems beneficial. We could also
> allow the external process to access the KV store in the JVM via the
> stdin/stdout protocol.

Yeah, I think allowing access to the KV store via stdin/stdout protocol makes the most sense. For example, to make a "get" request to the store, the task could write to stdout:

{"cmd": "kv_get", "store": "my-store", "key": "foo"}

to which Samza would respond by sending to stdin:

{"cmd": "kv_get_response", "store": "my-store", "key": "foo", "value": "bar"}

...or something of that sort.

> We'd also want to use an encoding scheme that
> doesn't have problems with separators (Hadoop Streaming has trouble with
> data that contains internal tabs and newlines). I think I'd prefer
> protobufs over JSON for performance reasons, but it could be configurable.

Protobuf would be an additional dependency, but I'd +1 it.

> So it seems like this is doable. There are some design questions but maybe
> a proof of concept would be a good next step. We'll be in touch if/when
> that happens. If anyone else wants to try it, we'd welcome that also :)

Would be awesome if you want to go ahead. Just say if you need any help.

Best,
Martin

Re: Stdin/stdout glue for external processes

[Dave Revell <da...@lytics.io>]

OP here, thanks to everyone for your thoughts. Replies inline:

In reply to Jakob, thanks for the pointer to SAMZA-18, that's good to know.
I probably won't hold my breath waiting for it though :)

In reply to Martin:

> Were you envisaging that such API calls could be made over a protocol
> that uses stdout and stdin as its transport? I'm sure it can be done, but
> the protocol would not be totally straightforward, as it would have to do
> things like match a request from the child process (on stdout) with a
> response to the API call (on stdin).

Samza currently has the nice property of having one message in flight at a
time per partition. I think we'd want to preserve that and avoid adding any
pipelining by having multiple messages in flight. So that means we could
use a synchronous protocol over stdin/stdout that doesn't need to match
requests to responses. Please correct me if I'm wrong on this, I'm still
very new to Samza.

In reply to Jay, I take your point about Hadoop Streaming. Adding
lifecycle/control messages to the protocol seems beneficial. We could also
allow the external process to access the KV store in the JVM via the
stdin/stdout protocol. We'd also want to use an encoding scheme that
doesn't have problems with separators (Hadoop Streaming has trouble with
data that contains internal tabs and newlines). I think I'd prefer
protobufs over JSON for performance reasons, but it could be configurable.

So it seems like this is doable. There are some design questions but maybe
a proof of concept would be a good next step. We'll be in touch if/when
that happens. If anyone else wants to try it, we'd welcome that also :)

Cheers,
Dave




On Tue, Mar 11, 2014 at 8:58 AM, Jay Kreps <ja...@gmail.com> wrote:

> Yes, I think all I am saying is that stdin/stdout aren't so bad. The
> mistake made by Hadoop streaming I think was to not specify a more detailed
> extensible control protocol for the client process. (or at least that
> seemed to be true a long-ass time ago when I last used Hadoop streaming).
> You would presumably model commands like commit as outputs and config and
> such as input and you would need so predetermined data format to read and
> write the topic/partition/key/value pairs. Sounds like Storm did a better
> job here.
>
> As you say this protocol needn't go over stdin/out but it is fairly cheap
> and I think unix domain sockets are not well supported in Java.
>
> As you say you definitely don't get the full power out of the box. For
> example the key-value store would have to be in the child process and all
> we could provide would be the backing changelog stream.
>
> Basically I am saying the maintenance burden of doing this seems low--it's
> just a simple samza job that manages a native subprocess and feeds it
> formatted input--so there is no harm in pursuing both approaches...
>
> -Jay
>
>
> On Mon, Mar 10, 2014 at 6:10 PM, Martin Kleppmann
> <mk...@linkedin.com>wrote:
>
> > If we want to make non-JVM languages first-class citizens in Samza (which
> > I think we should), they will need access to all the Samza APIs,
> including
> > reading and writing the key-value store and the like.
> >
> > Were you envisaging that such API calls could be made over a protocol
> that
> > uses stdout and stdin as its transport? I'm sure it can be done, but the
> > protocol would not be totally straightforward, as it would have to do
> > things like match a request from the child process (on stdout) with a
> > response to the API call (on stdin).
> >
> > Then there is a separate question of what transport the non-JVM process
> > uses to talk to the Samza container. stdout/stdin streams is an option,
> or
> > it could use Unix sockets, or TCP. Whatever method of transport is used,
> a
> > protocol will need to encode the Samza API calls in raw bytes.
> >
> > For reference, Storm uses a JSON-based protocol over stdin/stdout:
> > https://github.com/nathanmarz/storm/wiki/Multilang-protocol -- however,
> > it doesn't support all Storm features (
> > https://issues.apache.org/jira/browse/STORM-151).
> >
> > Martin
> >
> > On 10 Mar 2014, at 23:21, Jay Kreps <ja...@gmail.com> wrote:
> > > FWIW, I actually think the Hadoop streaming approach has some benefits.
> > It
> > > is less efficient then writing and embedding a C library but also much
> > much
> > > easier to implement and with less duplicate logic. I think we should be
> > > open to both of these--the streaming approach is so easy, it seems to
> me
> > > like there is not a huge downside to having that available.
> > >
> > > I think the mistake that Hadoop streaming might have made was
> > > over-simplifying the interaction with the client process. You probably
> > need
> > > a richer protocol than just the data (though I haven't thought this
> > > through).
> > >
> > > -Jay
> > >
> > >
> > > On Mon, Mar 10, 2014 at 12:26 PM, Jakob Homan <jg...@gmail.com>
> wrote:
> > >
> > >> Hey Dave-
> > >>   Thanks for taking a look at Samza.  No one in the community is
> > currently
> > >> working on this at the moment, to our knowledge.  SAMZA-18 (
> > >> https://issues.apache.org/jira/browse/SAMZA-18) has the beginnings
> of a
> > >> discussion about creating a single C library to help provide
> > multilanguage
> > >> support in Samza (which I believe would be accessible to Go as well).
> > >> There's currently no JIRA for Hadoop-style streaming, but one could
> > >> certainly be created and it would be something we'd be interested in.
> > >> Thanks,
> > >> Jakob
> > >>
> > >>
> > >>
> > >> On Mon, Mar 10, 2014 at 12:19 PM, Dave Revell <da...@lytics.io> wrote:
> > >>
> > >>> Hi all,
> > >>>
> > >>> We're considering using Samza for our high-throughput stream
> processing
> > >>> workload, but we don't want to rewrite all of our existing Go code.
> > We're
> > >>> considering writing something analogous to Hadoop Streaming, where
> the
> > >>> Samza consumer would start an external process and communicate with
> it
> > by
> > >>> passing protobufs via stdin/stdout. We like Samza's fault tolerance,
> > >> state
> > >>> management, and load balancing features and don't want to rewrite
> them.
> > >>>
> > >>> This possibility is mentioned in the documentation (
> > >>>
> > >>>
> > >>
> >
> http://samza.incubator.apache.org/learn/documentation/0.7.0/comparisons/storm.html
> > >>> ,
> > >>> search for "stdin") as something that might exist some day. My
> > >>> questions
> > >>> are:
> > >>>
> > >>> 1. Is anyone working on this, or planning to? I couldn't find any
> > related
> > >>> JIRAs.
> > >>> 2. Any advice for implementing this? Are there any challenges that
> > might
> > >>> not be obvious?
> > >>> 3. Should we try to merge this upstream?
> > >>>
> > >>> Thanks a bunch,
> > >>> Dave
> > >>>
> > >>
> >
> >
>

Re: Stdin/stdout glue for external processes

[Jay Kreps <ja...@gmail.com>]

Yes, I think all I am saying is that stdin/stdout aren't so bad. The
mistake made by Hadoop streaming I think was to not specify a more detailed
extensible control protocol for the client process. (or at least that
seemed to be true a long-ass time ago when I last used Hadoop streaming).
You would presumably model commands like commit as outputs and config and
such as input and you would need so predetermined data format to read and
write the topic/partition/key/value pairs. Sounds like Storm did a better
job here.

As you say this protocol needn't go over stdin/out but it is fairly cheap
and I think unix domain sockets are not well supported in Java.

As you say you definitely don't get the full power out of the box. For
example the key-value store would have to be in the child process and all
we could provide would be the backing changelog stream.

Basically I am saying the maintenance burden of doing this seems low--it's
just a simple samza job that manages a native subprocess and feeds it
formatted input--so there is no harm in pursuing both approaches...

-Jay


On Mon, Mar 10, 2014 at 6:10 PM, Martin Kleppmann
<mk...@linkedin.com>wrote:

> If we want to make non-JVM languages first-class citizens in Samza (which
> I think we should), they will need access to all the Samza APIs, including
> reading and writing the key-value store and the like.
>
> Were you envisaging that such API calls could be made over a protocol that
> uses stdout and stdin as its transport? I'm sure it can be done, but the
> protocol would not be totally straightforward, as it would have to do
> things like match a request from the child process (on stdout) with a
> response to the API call (on stdin).
>
> Then there is a separate question of what transport the non-JVM process
> uses to talk to the Samza container. stdout/stdin streams is an option, or
> it could use Unix sockets, or TCP. Whatever method of transport is used, a
> protocol will need to encode the Samza API calls in raw bytes.
>
> For reference, Storm uses a JSON-based protocol over stdin/stdout:
> https://github.com/nathanmarz/storm/wiki/Multilang-protocol -- however,
> it doesn't support all Storm features (
> https://issues.apache.org/jira/browse/STORM-151).
>
> Martin
>
> On 10 Mar 2014, at 23:21, Jay Kreps <ja...@gmail.com> wrote:
> > FWIW, I actually think the Hadoop streaming approach has some benefits.
> It
> > is less efficient then writing and embedding a C library but also much
> much
> > easier to implement and with less duplicate logic. I think we should be
> > open to both of these--the streaming approach is so easy, it seems to me
> > like there is not a huge downside to having that available.
> >
> > I think the mistake that Hadoop streaming might have made was
> > over-simplifying the interaction with the client process. You probably
> need
> > a richer protocol than just the data (though I haven't thought this
> > through).
> >
> > -Jay
> >
> >
> > On Mon, Mar 10, 2014 at 12:26 PM, Jakob Homan <jg...@gmail.com> wrote:
> >
> >> Hey Dave-
> >>   Thanks for taking a look at Samza.  No one in the community is
> currently
> >> working on this at the moment, to our knowledge.  SAMZA-18 (
> >> https://issues.apache.org/jira/browse/SAMZA-18) has the beginnings of a
> >> discussion about creating a single C library to help provide
> multilanguage
> >> support in Samza (which I believe would be accessible to Go as well).
> >> There's currently no JIRA for Hadoop-style streaming, but one could
> >> certainly be created and it would be something we'd be interested in.
> >> Thanks,
> >> Jakob
> >>
> >>
> >>
> >> On Mon, Mar 10, 2014 at 12:19 PM, Dave Revell <da...@lytics.io> wrote:
> >>
> >>> Hi all,
> >>>
> >>> We're considering using Samza for our high-throughput stream processing
> >>> workload, but we don't want to rewrite all of our existing Go code.
> We're
> >>> considering writing something analogous to Hadoop Streaming, where the
> >>> Samza consumer would start an external process and communicate with it
> by
> >>> passing protobufs via stdin/stdout. We like Samza's fault tolerance,
> >> state
> >>> management, and load balancing features and don't want to rewrite them.
> >>>
> >>> This possibility is mentioned in the documentation (
> >>>
> >>>
> >>
> http://samza.incubator.apache.org/learn/documentation/0.7.0/comparisons/storm.html
> >>> ,
> >>> search for "stdin") as something that might exist some day. My
> >>> questions
> >>> are:
> >>>
> >>> 1. Is anyone working on this, or planning to? I couldn't find any
> related
> >>> JIRAs.
> >>> 2. Any advice for implementing this? Are there any challenges that
> might
> >>> not be obvious?
> >>> 3. Should we try to merge this upstream?
> >>>
> >>> Thanks a bunch,
> >>> Dave
> >>>
> >>
>
>

Re: Stdin/stdout glue for external processes

[Martin Kleppmann <mk...@linkedin.com>]

If we want to make non-JVM languages first-class citizens in Samza (which I think we should), they will need access to all the Samza APIs, including reading and writing the key-value store and the like.

Were you envisaging that such API calls could be made over a protocol that uses stdout and stdin as its transport? I'm sure it can be done, but the protocol would not be totally straightforward, as it would have to do things like match a request from the child process (on stdout) with a response to the API call (on stdin).

Then there is a separate question of what transport the non-JVM process uses to talk to the Samza container. stdout/stdin streams is an option, or it could use Unix sockets, or TCP. Whatever method of transport is used, a protocol will need to encode the Samza API calls in raw bytes.

For reference, Storm uses a JSON-based protocol over stdin/stdout: https://github.com/nathanmarz/storm/wiki/Multilang-protocol -- however, it doesn't support all Storm features (https://issues.apache.org/jira/browse/STORM-151).

Martin

On 10 Mar 2014, at 23:21, Jay Kreps <ja...@gmail.com> wrote:
> FWIW, I actually think the Hadoop streaming approach has some benefits. It
> is less efficient then writing and embedding a C library but also much much
> easier to implement and with less duplicate logic. I think we should be
> open to both of these--the streaming approach is so easy, it seems to me
> like there is not a huge downside to having that available.
> 
> I think the mistake that Hadoop streaming might have made was
> over-simplifying the interaction with the client process. You probably need
> a richer protocol than just the data (though I haven't thought this
> through).
> 
> -Jay
> 
> 
> On Mon, Mar 10, 2014 at 12:26 PM, Jakob Homan <jg...@gmail.com> wrote:
> 
>> Hey Dave-
>>   Thanks for taking a look at Samza.  No one in the community is currently
>> working on this at the moment, to our knowledge.  SAMZA-18 (
>> https://issues.apache.org/jira/browse/SAMZA-18) has the beginnings of a
>> discussion about creating a single C library to help provide multilanguage
>> support in Samza (which I believe would be accessible to Go as well).
>> There's currently no JIRA for Hadoop-style streaming, but one could
>> certainly be created and it would be something we'd be interested in.
>> Thanks,
>> Jakob
>> 
>> 
>> 
>> On Mon, Mar 10, 2014 at 12:19 PM, Dave Revell <da...@lytics.io> wrote:
>> 
>>> Hi all,
>>> 
>>> We're considering using Samza for our high-throughput stream processing
>>> workload, but we don't want to rewrite all of our existing Go code. We're
>>> considering writing something analogous to Hadoop Streaming, where the
>>> Samza consumer would start an external process and communicate with it by
>>> passing protobufs via stdin/stdout. We like Samza's fault tolerance,
>> state
>>> management, and load balancing features and don't want to rewrite them.
>>> 
>>> This possibility is mentioned in the documentation (
>>> 
>>> 
>> http://samza.incubator.apache.org/learn/documentation/0.7.0/comparisons/storm.html
>>> ,
>>> search for "stdin") as something that might exist some day. My
>>> questions
>>> are:
>>> 
>>> 1. Is anyone working on this, or planning to? I couldn't find any related
>>> JIRAs.
>>> 2. Any advice for implementing this? Are there any challenges that might
>>> not be obvious?
>>> 3. Should we try to merge this upstream?
>>> 
>>> Thanks a bunch,
>>> Dave
>>> 
>> 

Re: Stdin/stdout glue for external processes

[Jay Kreps <ja...@gmail.com>]

FWIW, I actually think the Hadoop streaming approach has some benefits. It
is less efficient then writing and embedding a C library but also much much
easier to implement and with less duplicate logic. I think we should be
open to both of these--the streaming approach is so easy, it seems to me
like there is not a huge downside to having that available.

I think the mistake that Hadoop streaming might have made was
over-simplifying the interaction with the client process. You probably need
a richer protocol than just the data (though I haven't thought this
through).

-Jay


On Mon, Mar 10, 2014 at 12:26 PM, Jakob Homan <jg...@gmail.com> wrote:

> Hey Dave-
>    Thanks for taking a look at Samza.  No one in the community is currently
> working on this at the moment, to our knowledge.  SAMZA-18 (
> https://issues.apache.org/jira/browse/SAMZA-18) has the beginnings of a
> discussion about creating a single C library to help provide multilanguage
> support in Samza (which I believe would be accessible to Go as well).
> There's currently no JIRA for Hadoop-style streaming, but one could
> certainly be created and it would be something we'd be interested in.
> Thanks,
> Jakob
>
>
>
> On Mon, Mar 10, 2014 at 12:19 PM, Dave Revell <da...@lytics.io> wrote:
>
> > Hi all,
> >
> > We're considering using Samza for our high-throughput stream processing
> > workload, but we don't want to rewrite all of our existing Go code. We're
> > considering writing something analogous to Hadoop Streaming, where the
> > Samza consumer would start an external process and communicate with it by
> > passing protobufs via stdin/stdout. We like Samza's fault tolerance,
> state
> > management, and load balancing features and don't want to rewrite them.
> >
> > This possibility is mentioned in the documentation (
> >
> >
> http://samza.incubator.apache.org/learn/documentation/0.7.0/comparisons/storm.html
> > ,
> > search for "stdin") as something that might exist some day. My
> > questions
> > are:
> >
> > 1. Is anyone working on this, or planning to? I couldn't find any related
> > JIRAs.
> > 2. Any advice for implementing this? Are there any challenges that might
> > not be obvious?
> > 3. Should we try to merge this upstream?
> >
> > Thanks a bunch,
> > Dave
> >
>

Re: Stdin/stdout glue for external processes

[Jakob Homan <jg...@gmail.com>]

Hey Dave-
   Thanks for taking a look at Samza.  No one in the community is currently
working on this at the moment, to our knowledge.  SAMZA-18 (
https://issues.apache.org/jira/browse/SAMZA-18) has the beginnings of a
discussion about creating a single C library to help provide multilanguage
support in Samza (which I believe would be accessible to Go as well).
There's currently no JIRA for Hadoop-style streaming, but one could
certainly be created and it would be something we'd be interested in.
Thanks,
Jakob



On Mon, Mar 10, 2014 at 12:19 PM, Dave Revell <da...@lytics.io> wrote:

> Hi all,
>
> We're considering using Samza for our high-throughput stream processing
> workload, but we don't want to rewrite all of our existing Go code. We're
> considering writing something analogous to Hadoop Streaming, where the
> Samza consumer would start an external process and communicate with it by
> passing protobufs via stdin/stdout. We like Samza's fault tolerance, state
> management, and load balancing features and don't want to rewrite them.
>
> This possibility is mentioned in the documentation (
>
> http://samza.incubator.apache.org/learn/documentation/0.7.0/comparisons/storm.html
> ,
> search for "stdin") as something that might exist some day. My
> questions
> are:
>
> 1. Is anyone working on this, or planning to? I couldn't find any related
> JIRAs.
> 2. Any advice for implementing this? Are there any challenges that might
> not be obvious?
> 3. Should we try to merge this upstream?
>
> Thanks a bunch,
> Dave
>