You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@samza.apache.org by Elias Levy <fe...@gmail.com> on 2015/11/29 20:18:15 UTC
Executing Samza jobs natively in Kubernetes
I've been exploring Samza for stream processing as well as Kubernetes as a
container orchestration system and I wanted to be able to use one with the
other. The prospect of having to execute YARN either along side or on top
of Kubernetes did not appeal to me, so I developed a KubernetesJob
implementation of SamzaJob.
You can find the details at https://github.com/eliaslevy/samza_kubernetes,
but in summary KubernetesJob executes and generates a serialized JobModel.
Instead of interacting with Kubernetes directly to create the
SamzaContainers (as the YarnJob's SamzaApplicationMaster may do with the
YARN RM), it output a config YAML file that can be used to create the
SamzaContainers in Kubernetes by using Resource Controllers. For this you
require to package your job as a Docker image. You can reach the README at
the above repo for details.
A few observations:
It would be useful if SamzaContainer accepted the JobModel via an
environment variable. Right not it expects a URL to download it from. I
get around this by using a entry point script that copies the model from an
environment variable into a file, then passes a file URL to SamzaContainer.
SamzaContainer doesn't allow you to configure the JMX port. It selects a
port at random from the ephemeral range as it expects to execute in YARN
where a static port could result in a conflict. This is not the case in
Kubernetes where each Pod (i.e. SamzaContainer) is given its own IP
address.
This implementation doesn't provide a Samza dashboard, which in the YARN
implementation is hosted in the Application Master. There didn't seem to
be much value provided by the dashboard that is not already provided by the
Kubernetes tools for monitoring pods.
I've successfully executed the hello-samza jobs in Kubernetes:
$ kubectl get po
NAME READY STATUS RESTARTS AGE
kafka-1-jjh8n 1/1 Running 0 2d
kafka-2-buycp 1/1 Running 0 2d
kafka-3-tghkp 1/1 Running 0 2d
wikipedia-feed-0-4its2 1/1 Running 0 1d
wikipedia-parser-0-l0onv 1/1 Running 0 17h
wikipedia-parser-1-crrxh 1/1 Running 0 17h
wikipedia-parser-2-1c5nn 1/1 Running 0 17h
wikipedia-stats-0-3gaiu 1/1 Running 0 16h
wikipedia-stats-1-j5qlk 1/1 Running 0 16h
wikipedia-stats-2-2laos 1/1 Running 0 16h
zookeeper-1-1sb4a 1/1 Running 0 2d
zookeeper-2-dndk7 1/1 Running 0 2d
zookeeper-3-46n09 1/1 Running 0 2d
Finally, accessing services within the Kubernetes cluster from the outside
is quite cumbersome unless one uses an external load balancer. This makes
it difficult to bootstrap a job, as SamzaJob must connect to Zookeeper and
Kafka to find out the number of partitions on the topics it will subscribe
to, so it can assign them statically among the number of containers
requested.
Ideally Samza would operate along the lines of the Kafka high-level
consumer, which dynamically coordinate to allocate work among members of a
consumer group. This would do away with the new to execute SamzaJob a
priori to generate the JobModel to pass to the SamzaContainers. It would
also allow for dynamically changing the number of containers without having
the shutdown the job.
Re: Executing Samza jobs natively in Kubernetes
Posted by Jagadish Venkatraman <ja...@gmail.com>.
Hi Elias,
Thank you so much for your explanation. I'm looking into improving Samza
Standalone and your design-inputs were very useful.
Cheers,
Jagadish
On Mon, Dec 14, 2015 at 12:47 PM, Elias Levy <fe...@gmail.com>
wrote:
> On Mon, Dec 14, 2015 at 11:55 AM, Jagadish Venkatraman <
> jagadish1989@gmail.com> wrote:
>
> >
> > Thanks for the great work! This is super-helpful. Another cool feature is
> > that this implementation pushes alot of failure handling/restarting to
> the
> > cluster manager.
> >
> > I've some questions on the implementation.
> >
> > 1. It's my understanding that each container is a separate pod. When a
> pod
> > crashes, I assume that the *Kubelet* would bring back the job on the
> > **same**
> > node? (and we can re-use state since the volume is mounted as an
> *emptyDir*
> > volume) This is valuable for statefull jobs.
> >
>
> Correct. I create an emptyDir state volume on purpose, so that if the
> Docker container dies for some reason, the local state won't go away and
> will be available to the Samza container when it is restarted by Kubelet.
> I do the same thing for the logs, so you can fetch them and debug any
> issues that may be causing the container to fail.
>
>
>
> > 2. Also, the containerId and the jobModel are a part of the env. variable
> > passed to the pod. How do we ensure that - after a restart, the pods that
> > come up (potentially on a different host), have the same containerId?
> (Does
> > Kubernetes help in this by launching the process with the same env.
> > variables as the original run? )
> >
> > This behavior is key to ensure that we process all partitions with no 2
> > containers processing the same partition.
> >
>
> At the moment Kubernetes does not have a specific abstraction to model
> stateful services that have unique identities within the cluster (e.g. a
> Samza job, a Kafka cluster, a ZooKeeper ensemble), but it is easy to work
> past this limitation. You do so by creating one ReplicationController with
> a replica count of one per group member. In the case of a Samza job, my
> code creates a RC per SamzaContainer. Each RC has a slightly different Pod
> spec. Each Pod is parametrized a unique Samza container ID via the
> SAMZA_CONTAINER_ID environment variable. So the Kubernetes YAML the code
> outputs ensures there is only one RC per container ID, and the RC ensures
> there is only ever a single container running for its given configuration.
>
> Makes sense?
>
--
Jagadish V,
Graduate Student,
Department of Computer Science,
Stanford University
Re: Executing Samza jobs natively in Kubernetes
Posted by Elias Levy <fe...@gmail.com>.
On Mon, Dec 14, 2015 at 11:55 AM, Jagadish Venkatraman <
jagadish1989@gmail.com> wrote:
>
> Thanks for the great work! This is super-helpful. Another cool feature is
> that this implementation pushes alot of failure handling/restarting to the
> cluster manager.
>
> I've some questions on the implementation.
>
> 1. It's my understanding that each container is a separate pod. When a pod
> crashes, I assume that the *Kubelet* would bring back the job on the
> **same**
> node? (and we can re-use state since the volume is mounted as an *emptyDir*
> volume) This is valuable for statefull jobs.
>
Correct. I create an emptyDir state volume on purpose, so that if the
Docker container dies for some reason, the local state won't go away and
will be available to the Samza container when it is restarted by Kubelet.
I do the same thing for the logs, so you can fetch them and debug any
issues that may be causing the container to fail.
> 2. Also, the containerId and the jobModel are a part of the env. variable
> passed to the pod. How do we ensure that - after a restart, the pods that
> come up (potentially on a different host), have the same containerId? (Does
> Kubernetes help in this by launching the process with the same env.
> variables as the original run? )
>
> This behavior is key to ensure that we process all partitions with no 2
> containers processing the same partition.
>
At the moment Kubernetes does not have a specific abstraction to model
stateful services that have unique identities within the cluster (e.g. a
Samza job, a Kafka cluster, a ZooKeeper ensemble), but it is easy to work
past this limitation. You do so by creating one ReplicationController with
a replica count of one per group member. In the case of a Samza job, my
code creates a RC per SamzaContainer. Each RC has a slightly different Pod
spec. Each Pod is parametrized a unique Samza container ID via the
SAMZA_CONTAINER_ID environment variable. So the Kubernetes YAML the code
outputs ensures there is only one RC per container ID, and the RC ensures
there is only ever a single container running for its given configuration.
Makes sense?
Re: Executing Samza jobs natively in Kubernetes
Posted by Jagadish Venkatraman <ja...@gmail.com>.
Hey Elias,
Thanks for the great work! This is super-helpful. Another cool feature is
that this implementation pushes alot of failure handling/restarting to the
cluster manager.
I've some questions on the implementation.
1. It's my understanding that each container is a separate pod. When a pod
crashes, I assume that the *Kubelet* would bring back the job on the **same**
node? (and we can re-use state since the volume is mounted as an *emptyDir*
volume) This is valuable for statefull jobs.
2. Also, the containerId and the jobModel are a part of the env. variable
passed to the pod. How do we ensure that - after a restart, the pods that
come up (potentially on a different host), have the same containerId? (Does
Kubernetes help in this by launching the process with the same env.
variables as the original run? )
This behavior is key to ensure that we process all partitions with no 2
containers processing the same partition.
Cheers,
Jagadish
On Sun, Nov 29, 2015 at 11:18 AM, Elias Levy <fe...@gmail.com>
wrote:
> I've been exploring Samza for stream processing as well as Kubernetes as a
> container orchestration system and I wanted to be able to use one with the
> other. The prospect of having to execute YARN either along side or on top
> of Kubernetes did not appeal to me, so I developed a KubernetesJob
> implementation of SamzaJob.
>
> You can find the details at https://github.com/eliaslevy/samza_kubernetes,
> but in summary KubernetesJob executes and generates a serialized JobModel.
> Instead of interacting with Kubernetes directly to create the
> SamzaContainers (as the YarnJob's SamzaApplicationMaster may do with the
> YARN RM), it output a config YAML file that can be used to create the
> SamzaContainers in Kubernetes by using Resource Controllers. For this you
> require to package your job as a Docker image. You can reach the README at
> the above repo for details.
>
> A few observations:
>
> It would be useful if SamzaContainer accepted the JobModel via an
> environment variable. Right not it expects a URL to download it from. I
> get around this by using a entry point script that copies the model from an
> environment variable into a file, then passes a file URL to SamzaContainer.
>
> SamzaContainer doesn't allow you to configure the JMX port. It selects a
> port at random from the ephemeral range as it expects to execute in YARN
> where a static port could result in a conflict. This is not the case in
> Kubernetes where each Pod (i.e. SamzaContainer) is given its own IP
> address.
>
> This implementation doesn't provide a Samza dashboard, which in the YARN
> implementation is hosted in the Application Master. There didn't seem to
> be much value provided by the dashboard that is not already provided by the
> Kubernetes tools for monitoring pods.
>
> I've successfully executed the hello-samza jobs in Kubernetes:
>
> $ kubectl get po
> NAME READY STATUS RESTARTS AGE
> kafka-1-jjh8n 1/1 Running 0 2d
> kafka-2-buycp 1/1 Running 0 2d
> kafka-3-tghkp 1/1 Running 0 2d
> wikipedia-feed-0-4its2 1/1 Running 0 1d
> wikipedia-parser-0-l0onv 1/1 Running 0 17h
> wikipedia-parser-1-crrxh 1/1 Running 0 17h
> wikipedia-parser-2-1c5nn 1/1 Running 0 17h
> wikipedia-stats-0-3gaiu 1/1 Running 0 16h
> wikipedia-stats-1-j5qlk 1/1 Running 0 16h
> wikipedia-stats-2-2laos 1/1 Running 0 16h
> zookeeper-1-1sb4a 1/1 Running 0 2d
> zookeeper-2-dndk7 1/1 Running 0 2d
> zookeeper-3-46n09 1/1 Running 0 2d
>
>
> Finally, accessing services within the Kubernetes cluster from the outside
> is quite cumbersome unless one uses an external load balancer. This makes
> it difficult to bootstrap a job, as SamzaJob must connect to Zookeeper and
> Kafka to find out the number of partitions on the topics it will subscribe
> to, so it can assign them statically among the number of containers
> requested.
>
> Ideally Samza would operate along the lines of the Kafka high-level
> consumer, which dynamically coordinate to allocate work among members of a
> consumer group. This would do away with the new to execute SamzaJob a
> priori to generate the JobModel to pass to the SamzaContainers. It would
> also allow for dynamically changing the number of containers without having
> the shutdown the job.
>
--
Jagadish V,
Graduate Student,
Department of Computer Science,
Stanford University
Re: Executing Samza jobs natively in Kubernetes
Posted by Elias Levy <fe...@gmail.com>.
On Mon, Nov 30, 2015 at 12:31 PM, Jakob Homan <jg...@gmail.com> wrote:
> This is awesome work.
Thanks. Feel like a hack, but it got the job done.
> Would be interested in opening JIRAs for the
> changes you need so we can start to process them?
>
I've opened:
- SAMZA-824 <https://issues.apache.org/jira/browse/SAMZA-824>:
SamzaContainer should optionally accept its config via the environment.
- SAMZA-825 <https://issues.apache.org/jira/browse/SAMZA-825>:
SamzaContainer
JMX port should be configurable
As for my last comment, I think its already covered by SAMZA-348
<https://issues.apache.org/jira/browse/SAMZA-348>, although I think you
expressed it better in your comment on SAMZA-247
<https://issues.apache.org/jira/browse/SAMZA-237>.
Re: Executing Samza jobs natively in Kubernetes
Posted by Yi Pan <ni...@gmail.com>.
Hi, Elias,
Thanks a lot to put up the patch for the simple job running in Kubernetes!
As Kartik mentioned, that is well aligned w/ our goal to make Samza job
launching easier. I am glad that we actually share a lot of common ideas
from independent minds. Let me try to give my opinions on this:
1. My view on the role of cluster management systems (i.e. YARN, Mesos,
Docker, Kubernetes, etc.): the cluster management system should be helping
in the following aspects:
a. help to allocate/de-allocate "physical" resource to run the
standalone SamzaContainer processes
i. this *must* help in initial allocation of physical resources (i.e.
YARN, Mesos container, Docker, etc)
ii. this *must* help in re-allocating the physical resource when
trying to re-launch the failed standalone SamzaContainer process
iii. this *should* help in dynamically allocate/de-allocate physical
resources to re-size the Samza job
b. help to monitor the liveness of the standalone SamzaContainer
processes
c. help to re-launch the failed standalone SamzaContainer processes
within a reasonable limits. That should be done before the
re-balancing-like behavior in SAMZA-516 is triggered.
2. The following functions should be in Samza itself, not relying on the
underlying cluster management systems:
a. Partition assignment: this would include the logic of grouping input
topic partitions in a task, and assign tasks to each SamzaContainer
processes. It should be pluggable/configurable to allow static assignment
w/o coordination (e.g. SAMZA-41), or some coordination implementation (e.g.
SAMZA-516)
b. Decision on re-sizing the job: this would include the logic Kartik
and you have mentioned as item (2), that Samza itself needs to determine to
reduce/increase the number of SamzaContainer processes needed in a job
(e.g. rebalancing as in SAMZA-516 is part of this). This would require
i. An abstract interface to the cluster management system to request
physical resource and launch additional SamzaContainer processes; and
shutdown some SamzaContainer processes and release some physical resources.
ii. Arguably, this would require coordination to elect a leader
(JobCoordinator) for decision making per Samza job.
c. More advanced features of container launching that depends on
application state (e.g. Samza application state store). One example is
host-affinity that allows to match each SamzaContainer to the physical
resource for potential state-reuse. I would argue that most of these
features would require JobCoordinator as well.
And some of my cents on the technical details below:
On Tue, Dec 1, 2015 at 9:14 AM, Elias Levy <fe...@gmail.com>
wrote:
> That said, yes, I do believe dynamic assignment should be implemented. Not
> because of fault-tolerance, which I think is the domain of whatever
> container execution system is used to implement the job, but to enable
> scaling up and down of a job or the dynamic addition and removal of
> partitions without having to restart the job.
>
Totally agreed.
> One aspect of the StandaloneJob proposal that was somewhat off-putting was
> that the rebalancing was performed by shifting SamzaContainers around
> rather than shifting partitions among containers. That adds an extra layer
> of conceptual complexity. I understand that from an implementation
> perspective the easiest thing to do is to shift SamzaContainers around, as
> they appear to be immutable after the job starts and changing that may
> require a lot of work, but conceptually it seems to make more sense to
> shift partitions among SamzaContainers and execute a single SamzaContainer
> per "container" (e.g. YARN container, Docker container, etc). Then again,
> that is an implementation detail that would be largely hidden from the job
> developer, although it would be exposed to the job admin, as the number of
> SamzaContainers would become his parallelism limit.
>
> Yes, I agree w/ the reasoning here. As you said, there are some
implementation details involved here. But it should not matter to the users.
>
> > > I submit that there is also a need for a simpler type of job, lets call
> > it
> > > SimpleJob. This would be a job where, at least until dynamic
> > configuration
> > > is implemented, the JobCoordinator is executed once a priori to
> generate
> > a
> > > JobModel, the user can then configure the JobModel in the containers'
> > > environment, and execute the containers as he sees fit.
>
Yes. I agree w/ the idea that we probably need to support this model as
well. Just one technical issue: passing JobModel via container's
environment may cause the same issue that motivated us to implement the
CoordinatorStream in 0.10 (e.g. SAMZA-333). One potential solution is that
each SamzaContainer also reads from the CoordinatorStream and builds up the
JobModel locally during bootstrap, which eliminates the need of a single
JobCoordinator's HTTP API executed a priori. More details can be discussed
on this topic later.
> > >
> > > Failure handling and container monitoring is then left as something to
> be
> > > handled by whatever system the user chooses to execute the containers.
> > >
>
> > This model can then be used with many of the existing container
> > > orchestration systems, whether it be Kubernetes, Docker Swarm, Amazon
> EC2
> > > Container Service, CoreOS Fleet, Helios, Marathon, Nomad, etc.
> > >
> > > In fact, KubernetesJob is essentially this SimpleJob proposal, except
> > that
> > > it outputs a Kubernetes config file instead of only the JobModel.
> > >
>
Totally agree the above points.
Thanks a lot!
-Yi
Re: Executing Samza jobs natively in Kubernetes
Posted by Elias Levy <fe...@gmail.com>.
I was too critical in my last comment on StandaloneJob. More specifically,
StandaloneJob did not aid much on my implementation, which is based on the
current 0.9.1 model of static a priori partition assignment. Within that
model all that is needed is a command that takes in the job config and
spits out the serialized JobModel, and an easy method to pass the JobModel
to SamzaContainer, which can then be used to start SamzaContainers within
whatever container execution framework you've chosen.
That said, yes, I do believe dynamic assignment should be implemented. Not
because of fault-tolerance, which I think is the domain of whatever
container execution system is used to implement the job, but to enable
scaling up and down of a job or the dynamic addition and removal of
partitions without having to restart the job.
Dynamic assignment would also null the need in many system for a JobRunner
that requires access to the cluster services, making it easier to bootstrap
a job using only a config file and a number of immutable SamzaContainers
that implement the job. I think we all agree that Samza's current job
bootstrapping is convoluted.
I agree with you that work on dynamic assignment is additive on work on a
simpler job that is statically configured on startup.
One aspect of the StandaloneJob proposal that was somewhat off-putting was
that the rebalancing was performed by shifting SamzaContainers around
rather than shifting partitions among containers. That adds an extra layer
of conceptual complexity. I understand that from an implementation
perspective the easiest thing to do is to shift SamzaContainers around, as
they appear to be immutable after the job starts and changing that may
require a lot of work, but conceptually it seems to make more sense to
shift partitions among SamzaContainers and execute a single SamzaContainer
per "container" (e.g. YARN container, Docker container, etc). Then again,
that is an implementation detail that would be largely hidden from the job
developer, although it would be exposed to the job admin, as the number of
SamzaContainers would become his parallelism limit.
On Mon, Nov 30, 2015 at 10:14 PM, Kartik Paramasivam <
kparamasivam@linkedin.com.invalid> wrote:
> Thanks for starting this discussion and great to see the integration with
> Kubernetes. It is very timely because we(at LinkedIn) have recently
> started to refocus on Samza-516 and were thinking of kicking off a similar
> discussion soon.
>
> I am in agreement with you that there are two possible levels to the
> standalone Samza (or rather pluggable cluster manager) implementation.
>
> 1. Let the cluster management take care of fault-tolerance for "container"
> failures.
> - i.e. if the machine for a container dies, we leave it upto the cluster
> manager of choice to start up the container on another machine. No attempt
> is made to redistribute the partitions among the existing containers.
>
> 2. The partition to container mapping gets dynamically adjusted to deal
> with "container/node" death.
> - your earlier comment of making samza act more like the kafka consumer
> actually is more inline with this model. This is also what Chris had
> implemented in Samza 516.
>
> The way I see if (2) should be an additive (and possibly opt-in)
> improvement over 1.
>
> In your last email you seem to indicate that the additional functionality
> of dynamic partition/container re-assignment on container/node failures
> that is implemented in Samza-516 is problemmatic for Kubernetes
> integration. Could you provide some more details on why this doesn't work
> for you ??
>
> From my standpoint, there is another reason it might be attractive to
> separate (2) from (1) : engineering expediency. i.e. (2) adds a lot more
> complexity and I feel it will take much longer to stabilize. This is more
> true now that we have host-affinity also fully implemented in Samza 0.10.
> (1) on the other hand is easier to stabilize and still covers most of the
> scenarios.
>
> Since we are on the subject, it would be good to further elaborate on (1)
> .. Here is how I think of (1).
> Basically when the job starts for the first time, the system would
> dynamically distribute all the partitions across the configured number of
> containers. So the only thing that is statically configured here is the
> number of containers (same as how things are today). This mapping would
> be stored durably using the same mechanism that has been introduced in
> Samza 0.10 (coordinator stream).
>
> One further addition to this model would be to deal with explicit container
> additions and removals. For e.g. if a new container is explicitly added or
> removed via a config change, the system would change the partition to
> container mapping (and persist it durably back in the coordinator stream).
>
> Thanks
> Kartik
>
>
>
> On Mon, Nov 30, 2015 at 5:54 PM, Elias Levy <fe...@gmail.com>
> wrote:
>
> > BTW, I reviewed the StandaloneJob proposal in SAMZA-516 thinking it could
> > have been useful in running Samza in Kubernetes and I was disappointed to
> > find out it was not the case. StandaloneJob goes a step beyond the most
> > basic job implementation as it attempts to handle failures through
> > rebalances managed by the JobCoordinator which is elected via ZoKeeper.
> >
> > I submit that there is also a need for a simpler type of job, lets call
> it
> > SimpleJob. This would be a job where, at least until dynamic
> configuration
> > is implemented, the JobCoordinator is executed once a priori to generate
> a
> > JobModel, the user can then configure the JobModel in the containers'
> > environment, and execute the containers as he sees fit.
> >
> > Failure handling and container monitoring is then left as something to be
> > handled by whatever system the user chooses to execute the containers.
> >
> > This model can then be used with many of the existing container
> > orchestration systems, whether it be Kubernetes, Docker Swarm, Amazon EC2
> > Container Service, CoreOS Fleet, Helios, Marathon, Nomad, etc.
> >
> > In fact, KubernetesJob is essentially this SimpleJob proposal, except
> that
> > it outputs a Kubernetes config file instead of only the JobModel.
> >
> >
> >
> > On Mon, Nov 30, 2015 at 12:31 PM, Jakob Homan <jg...@gmail.com> wrote:
> >
> > > This is awesome work. Would be interested in opening JIRAs for the
> > > changes you need so we can start to process them?
> >
>
Re: Executing Samza jobs natively in Kubernetes
Posted by Kartik Paramasivam <kp...@linkedin.com.INVALID>.
Thanks for starting this discussion and great to see the integration with
Kubernetes. It is very timely because we(at LinkedIn) have recently
started to refocus on Samza-516 and were thinking of kicking off a similar
discussion soon.
I am in agreement with you that there are two possible levels to the
standalone Samza (or rather pluggable cluster manager) implementation.
1. Let the cluster management take care of fault-tolerance for "container"
failures.
- i.e. if the machine for a container dies, we leave it upto the cluster
manager of choice to start up the container on another machine. No attempt
is made to redistribute the partitions among the existing containers.
2. The partition to container mapping gets dynamically adjusted to deal
with "container/node" death.
- your earlier comment of making samza act more like the kafka consumer
actually is more inline with this model. This is also what Chris had
implemented in Samza 516.
The way I see if (2) should be an additive (and possibly opt-in)
improvement over 1.
In your last email you seem to indicate that the additional functionality
of dynamic partition/container re-assignment on container/node failures
that is implemented in Samza-516 is problemmatic for Kubernetes
integration. Could you provide some more details on why this doesn't work
for you ??
>From my standpoint, there is another reason it might be attractive to
separate (2) from (1) : engineering expediency. i.e. (2) adds a lot more
complexity and I feel it will take much longer to stabilize. This is more
true now that we have host-affinity also fully implemented in Samza 0.10.
(1) on the other hand is easier to stabilize and still covers most of the
scenarios.
Since we are on the subject, it would be good to further elaborate on (1)
.. Here is how I think of (1).
Basically when the job starts for the first time, the system would
dynamically distribute all the partitions across the configured number of
containers. So the only thing that is statically configured here is the
number of containers (same as how things are today). This mapping would
be stored durably using the same mechanism that has been introduced in
Samza 0.10 (coordinator stream).
One further addition to this model would be to deal with explicit container
additions and removals. For e.g. if a new container is explicitly added or
removed via a config change, the system would change the partition to
container mapping (and persist it durably back in the coordinator stream).
Thanks
Kartik
On Mon, Nov 30, 2015 at 5:54 PM, Elias Levy <fe...@gmail.com>
wrote:
> BTW, I reviewed the StandaloneJob proposal in SAMZA-516 thinking it could
> have been useful in running Samza in Kubernetes and I was disappointed to
> find out it was not the case. StandaloneJob goes a step beyond the most
> basic job implementation as it attempts to handle failures through
> rebalances managed by the JobCoordinator which is elected via ZoKeeper.
>
> I submit that there is also a need for a simpler type of job, lets call it
> SimpleJob. This would be a job where, at least until dynamic configuration
> is implemented, the JobCoordinator is executed once a priori to generate a
> JobModel, the user can then configure the JobModel in the containers'
> environment, and execute the containers as he sees fit.
>
> Failure handling and container monitoring is then left as something to be
> handled by whatever system the user chooses to execute the containers.
>
> This model can then be used with many of the existing container
> orchestration systems, whether it be Kubernetes, Docker Swarm, Amazon EC2
> Container Service, CoreOS Fleet, Helios, Marathon, Nomad, etc.
>
> In fact, KubernetesJob is essentially this SimpleJob proposal, except that
> it outputs a Kubernetes config file instead of only the JobModel.
>
>
>
> On Mon, Nov 30, 2015 at 12:31 PM, Jakob Homan <jg...@gmail.com> wrote:
>
> > This is awesome work. Would be interested in opening JIRAs for the
> > changes you need so we can start to process them?
>
Re: Executing Samza jobs natively in Kubernetes
Posted by Elias Levy <fe...@gmail.com>.
BTW, I reviewed the StandaloneJob proposal in SAMZA-516 thinking it could
have been useful in running Samza in Kubernetes and I was disappointed to
find out it was not the case. StandaloneJob goes a step beyond the most
basic job implementation as it attempts to handle failures through
rebalances managed by the JobCoordinator which is elected via ZoKeeper.
I submit that there is also a need for a simpler type of job, lets call it
SimpleJob. This would be a job where, at least until dynamic configuration
is implemented, the JobCoordinator is executed once a priori to generate a
JobModel, the user can then configure the JobModel in the containers'
environment, and execute the containers as he sees fit.
Failure handling and container monitoring is then left as something to be
handled by whatever system the user chooses to execute the containers.
This model can then be used with many of the existing container
orchestration systems, whether it be Kubernetes, Docker Swarm, Amazon EC2
Container Service, CoreOS Fleet, Helios, Marathon, Nomad, etc.
In fact, KubernetesJob is essentially this SimpleJob proposal, except that
it outputs a Kubernetes config file instead of only the JobModel.
On Mon, Nov 30, 2015 at 12:31 PM, Jakob Homan <jg...@gmail.com> wrote:
> This is awesome work. Would be interested in opening JIRAs for the
> changes you need so we can start to process them?
Re: Executing Samza jobs natively in Kubernetes
Posted by Jakob Homan <jg...@gmail.com>.
Hey Elias-
This is awesome work. Would be interested in opening JIRAs for the
changes you need so we can start to process them?
Thanks,
Jakob
On 30 November 2015 at 12:18, Roger Hoover <ro...@gmail.com> wrote:
> Awesome. Thanks.
>
> On Sun, Nov 29, 2015 at 3:25 PM, Elias Levy <fe...@gmail.com>
> wrote:
>
>> Roger,
>>
>> You are welcomed. If you want to experiment, you can use my hello samza
>> <https://hub.docker.com/r/elevy/hello-samza/> Docker image.
>>
>> On Sun, Nov 29, 2015 at 12:19 PM, Roger Hoover <ro...@gmail.com>
>> wrote:
>>
>> > Elias,
>> >
>> > I would also love to be able to deploy Samza on Kubernetes with dynamic
>> > task management. Thanks for sharing this. It may be a good interim
>> > solution.
>> >
>> > Roger
>> >
>> > On Sun, Nov 29, 2015 at 11:18 AM, Elias Levy <
>> fearsome.lucidity@gmail.com>
>> > wrote:
>> >
>> > > I've been exploring Samza for stream processing as well as Kubernetes
>> as
>> > a
>> > > container orchestration system and I wanted to be able to use one with
>> > the
>> > > other. The prospect of having to execute YARN either along side or on
>> > top
>> > > of Kubernetes did not appeal to me, so I developed a KubernetesJob
>> > > implementation of SamzaJob.
>> > >
>> > > You can find the details at
>> > https://github.com/eliaslevy/samza_kubernetes,
>> > > but in summary KubernetesJob executes and generates a serialized
>> > JobModel.
>> > > Instead of interacting with Kubernetes directly to create the
>> > > SamzaContainers (as the YarnJob's SamzaApplicationMaster may do with
>> the
>> > > YARN RM), it output a config YAML file that can be used to create the
>> > > SamzaContainers in Kubernetes by using Resource Controllers. For this
>> > you
>> > > require to package your job as a Docker image. You can reach the
>> README
>> > at
>> > > the above repo for details.
>> > >
>> > > A few observations:
>> > >
>> > > It would be useful if SamzaContainer accepted the JobModel via an
>> > > environment variable. Right not it expects a URL to download it
>> from. I
>> > > get around this by using a entry point script that copies the model
>> from
>> > an
>> > > environment variable into a file, then passes a file URL to
>> > SamzaContainer.
>> > >
>> > > SamzaContainer doesn't allow you to configure the JMX port. It
>> selects a
>> > > port at random from the ephemeral range as it expects to execute in
>> YARN
>> > > where a static port could result in a conflict. This is not the case
>> in
>> > > Kubernetes where each Pod (i.e. SamzaContainer) is given its own IP
>> > > address.
>> > >
>> > > This implementation doesn't provide a Samza dashboard, which in the
>> YARN
>> > > implementation is hosted in the Application Master. There didn't seem
>> to
>> > > be much value provided by the dashboard that is not already provided by
>> > the
>> > > Kubernetes tools for monitoring pods.
>> > >
>> > > I've successfully executed the hello-samza jobs in Kubernetes:
>> > >
>> > > $ kubectl get po
>> > > NAME READY STATUS RESTARTS AGE
>> > > kafka-1-jjh8n 1/1 Running 0 2d
>> > > kafka-2-buycp 1/1 Running 0 2d
>> > > kafka-3-tghkp 1/1 Running 0 2d
>> > > wikipedia-feed-0-4its2 1/1 Running 0 1d
>> > > wikipedia-parser-0-l0onv 1/1 Running 0 17h
>> > > wikipedia-parser-1-crrxh 1/1 Running 0 17h
>> > > wikipedia-parser-2-1c5nn 1/1 Running 0 17h
>> > > wikipedia-stats-0-3gaiu 1/1 Running 0 16h
>> > > wikipedia-stats-1-j5qlk 1/1 Running 0 16h
>> > > wikipedia-stats-2-2laos 1/1 Running 0 16h
>> > > zookeeper-1-1sb4a 1/1 Running 0 2d
>> > > zookeeper-2-dndk7 1/1 Running 0 2d
>> > > zookeeper-3-46n09 1/1 Running 0 2d
>> > >
>> > >
>> > > Finally, accessing services within the Kubernetes cluster from the
>> > outside
>> > > is quite cumbersome unless one uses an external load balancer. This
>> > makes
>> > > it difficult to bootstrap a job, as SamzaJob must connect to Zookeeper
>> > and
>> > > Kafka to find out the number of partitions on the topics it will
>> > subscribe
>> > > to, so it can assign them statically among the number of containers
>> > > requested.
>> > >
>> > > Ideally Samza would operate along the lines of the Kafka high-level
>> > > consumer, which dynamically coordinate to allocate work among members
>> of
>> > a
>> > > consumer group. This would do away with the new to execute SamzaJob a
>> > > priori to generate the JobModel to pass to the SamzaContainers. It
>> would
>> > > also allow for dynamically changing the number of containers without
>> > having
>> > > the shutdown the job.
>> > >
>> >
>>
Re: Executing Samza jobs natively in Kubernetes
Posted by Roger Hoover <ro...@gmail.com>.
Awesome. Thanks.
On Sun, Nov 29, 2015 at 3:25 PM, Elias Levy <fe...@gmail.com>
wrote:
> Roger,
>
> You are welcomed. If you want to experiment, you can use my hello samza
> <https://hub.docker.com/r/elevy/hello-samza/> Docker image.
>
> On Sun, Nov 29, 2015 at 12:19 PM, Roger Hoover <ro...@gmail.com>
> wrote:
>
> > Elias,
> >
> > I would also love to be able to deploy Samza on Kubernetes with dynamic
> > task management. Thanks for sharing this. It may be a good interim
> > solution.
> >
> > Roger
> >
> > On Sun, Nov 29, 2015 at 11:18 AM, Elias Levy <
> fearsome.lucidity@gmail.com>
> > wrote:
> >
> > > I've been exploring Samza for stream processing as well as Kubernetes
> as
> > a
> > > container orchestration system and I wanted to be able to use one with
> > the
> > > other. The prospect of having to execute YARN either along side or on
> > top
> > > of Kubernetes did not appeal to me, so I developed a KubernetesJob
> > > implementation of SamzaJob.
> > >
> > > You can find the details at
> > https://github.com/eliaslevy/samza_kubernetes,
> > > but in summary KubernetesJob executes and generates a serialized
> > JobModel.
> > > Instead of interacting with Kubernetes directly to create the
> > > SamzaContainers (as the YarnJob's SamzaApplicationMaster may do with
> the
> > > YARN RM), it output a config YAML file that can be used to create the
> > > SamzaContainers in Kubernetes by using Resource Controllers. For this
> > you
> > > require to package your job as a Docker image. You can reach the
> README
> > at
> > > the above repo for details.
> > >
> > > A few observations:
> > >
> > > It would be useful if SamzaContainer accepted the JobModel via an
> > > environment variable. Right not it expects a URL to download it
> from. I
> > > get around this by using a entry point script that copies the model
> from
> > an
> > > environment variable into a file, then passes a file URL to
> > SamzaContainer.
> > >
> > > SamzaContainer doesn't allow you to configure the JMX port. It
> selects a
> > > port at random from the ephemeral range as it expects to execute in
> YARN
> > > where a static port could result in a conflict. This is not the case
> in
> > > Kubernetes where each Pod (i.e. SamzaContainer) is given its own IP
> > > address.
> > >
> > > This implementation doesn't provide a Samza dashboard, which in the
> YARN
> > > implementation is hosted in the Application Master. There didn't seem
> to
> > > be much value provided by the dashboard that is not already provided by
> > the
> > > Kubernetes tools for monitoring pods.
> > >
> > > I've successfully executed the hello-samza jobs in Kubernetes:
> > >
> > > $ kubectl get po
> > > NAME READY STATUS RESTARTS AGE
> > > kafka-1-jjh8n 1/1 Running 0 2d
> > > kafka-2-buycp 1/1 Running 0 2d
> > > kafka-3-tghkp 1/1 Running 0 2d
> > > wikipedia-feed-0-4its2 1/1 Running 0 1d
> > > wikipedia-parser-0-l0onv 1/1 Running 0 17h
> > > wikipedia-parser-1-crrxh 1/1 Running 0 17h
> > > wikipedia-parser-2-1c5nn 1/1 Running 0 17h
> > > wikipedia-stats-0-3gaiu 1/1 Running 0 16h
> > > wikipedia-stats-1-j5qlk 1/1 Running 0 16h
> > > wikipedia-stats-2-2laos 1/1 Running 0 16h
> > > zookeeper-1-1sb4a 1/1 Running 0 2d
> > > zookeeper-2-dndk7 1/1 Running 0 2d
> > > zookeeper-3-46n09 1/1 Running 0 2d
> > >
> > >
> > > Finally, accessing services within the Kubernetes cluster from the
> > outside
> > > is quite cumbersome unless one uses an external load balancer. This
> > makes
> > > it difficult to bootstrap a job, as SamzaJob must connect to Zookeeper
> > and
> > > Kafka to find out the number of partitions on the topics it will
> > subscribe
> > > to, so it can assign them statically among the number of containers
> > > requested.
> > >
> > > Ideally Samza would operate along the lines of the Kafka high-level
> > > consumer, which dynamically coordinate to allocate work among members
> of
> > a
> > > consumer group. This would do away with the new to execute SamzaJob a
> > > priori to generate the JobModel to pass to the SamzaContainers. It
> would
> > > also allow for dynamically changing the number of containers without
> > having
> > > the shutdown the job.
> > >
> >
>
Re: Executing Samza jobs natively in Kubernetes
Posted by Elias Levy <fe...@gmail.com>.
Roger,
You are welcomed. If you want to experiment, you can use my hello samza
<https://hub.docker.com/r/elevy/hello-samza/> Docker image.
On Sun, Nov 29, 2015 at 12:19 PM, Roger Hoover <ro...@gmail.com>
wrote:
> Elias,
>
> I would also love to be able to deploy Samza on Kubernetes with dynamic
> task management. Thanks for sharing this. It may be a good interim
> solution.
>
> Roger
>
> On Sun, Nov 29, 2015 at 11:18 AM, Elias Levy <fe...@gmail.com>
> wrote:
>
> > I've been exploring Samza for stream processing as well as Kubernetes as
> a
> > container orchestration system and I wanted to be able to use one with
> the
> > other. The prospect of having to execute YARN either along side or on
> top
> > of Kubernetes did not appeal to me, so I developed a KubernetesJob
> > implementation of SamzaJob.
> >
> > You can find the details at
> https://github.com/eliaslevy/samza_kubernetes,
> > but in summary KubernetesJob executes and generates a serialized
> JobModel.
> > Instead of interacting with Kubernetes directly to create the
> > SamzaContainers (as the YarnJob's SamzaApplicationMaster may do with the
> > YARN RM), it output a config YAML file that can be used to create the
> > SamzaContainers in Kubernetes by using Resource Controllers. For this
> you
> > require to package your job as a Docker image. You can reach the README
> at
> > the above repo for details.
> >
> > A few observations:
> >
> > It would be useful if SamzaContainer accepted the JobModel via an
> > environment variable. Right not it expects a URL to download it from. I
> > get around this by using a entry point script that copies the model from
> an
> > environment variable into a file, then passes a file URL to
> SamzaContainer.
> >
> > SamzaContainer doesn't allow you to configure the JMX port. It selects a
> > port at random from the ephemeral range as it expects to execute in YARN
> > where a static port could result in a conflict. This is not the case in
> > Kubernetes where each Pod (i.e. SamzaContainer) is given its own IP
> > address.
> >
> > This implementation doesn't provide a Samza dashboard, which in the YARN
> > implementation is hosted in the Application Master. There didn't seem to
> > be much value provided by the dashboard that is not already provided by
> the
> > Kubernetes tools for monitoring pods.
> >
> > I've successfully executed the hello-samza jobs in Kubernetes:
> >
> > $ kubectl get po
> > NAME READY STATUS RESTARTS AGE
> > kafka-1-jjh8n 1/1 Running 0 2d
> > kafka-2-buycp 1/1 Running 0 2d
> > kafka-3-tghkp 1/1 Running 0 2d
> > wikipedia-feed-0-4its2 1/1 Running 0 1d
> > wikipedia-parser-0-l0onv 1/1 Running 0 17h
> > wikipedia-parser-1-crrxh 1/1 Running 0 17h
> > wikipedia-parser-2-1c5nn 1/1 Running 0 17h
> > wikipedia-stats-0-3gaiu 1/1 Running 0 16h
> > wikipedia-stats-1-j5qlk 1/1 Running 0 16h
> > wikipedia-stats-2-2laos 1/1 Running 0 16h
> > zookeeper-1-1sb4a 1/1 Running 0 2d
> > zookeeper-2-dndk7 1/1 Running 0 2d
> > zookeeper-3-46n09 1/1 Running 0 2d
> >
> >
> > Finally, accessing services within the Kubernetes cluster from the
> outside
> > is quite cumbersome unless one uses an external load balancer. This
> makes
> > it difficult to bootstrap a job, as SamzaJob must connect to Zookeeper
> and
> > Kafka to find out the number of partitions on the topics it will
> subscribe
> > to, so it can assign them statically among the number of containers
> > requested.
> >
> > Ideally Samza would operate along the lines of the Kafka high-level
> > consumer, which dynamically coordinate to allocate work among members of
> a
> > consumer group. This would do away with the new to execute SamzaJob a
> > priori to generate the JobModel to pass to the SamzaContainers. It would
> > also allow for dynamically changing the number of containers without
> having
> > the shutdown the job.
> >
>
Re: Executing Samza jobs natively in Kubernetes
Posted by Roger Hoover <ro...@gmail.com>.
Elias,
I would also love to be able to deploy Samza on Kubernetes with dynamic
task management. Thanks for sharing this. It may be a good interim
solution.
Roger
On Sun, Nov 29, 2015 at 11:18 AM, Elias Levy <fe...@gmail.com>
wrote:
> I've been exploring Samza for stream processing as well as Kubernetes as a
> container orchestration system and I wanted to be able to use one with the
> other. The prospect of having to execute YARN either along side or on top
> of Kubernetes did not appeal to me, so I developed a KubernetesJob
> implementation of SamzaJob.
>
> You can find the details at https://github.com/eliaslevy/samza_kubernetes,
> but in summary KubernetesJob executes and generates a serialized JobModel.
> Instead of interacting with Kubernetes directly to create the
> SamzaContainers (as the YarnJob's SamzaApplicationMaster may do with the
> YARN RM), it output a config YAML file that can be used to create the
> SamzaContainers in Kubernetes by using Resource Controllers. For this you
> require to package your job as a Docker image. You can reach the README at
> the above repo for details.
>
> A few observations:
>
> It would be useful if SamzaContainer accepted the JobModel via an
> environment variable. Right not it expects a URL to download it from. I
> get around this by using a entry point script that copies the model from an
> environment variable into a file, then passes a file URL to SamzaContainer.
>
> SamzaContainer doesn't allow you to configure the JMX port. It selects a
> port at random from the ephemeral range as it expects to execute in YARN
> where a static port could result in a conflict. This is not the case in
> Kubernetes where each Pod (i.e. SamzaContainer) is given its own IP
> address.
>
> This implementation doesn't provide a Samza dashboard, which in the YARN
> implementation is hosted in the Application Master. There didn't seem to
> be much value provided by the dashboard that is not already provided by the
> Kubernetes tools for monitoring pods.
>
> I've successfully executed the hello-samza jobs in Kubernetes:
>
> $ kubectl get po
> NAME READY STATUS RESTARTS AGE
> kafka-1-jjh8n 1/1 Running 0 2d
> kafka-2-buycp 1/1 Running 0 2d
> kafka-3-tghkp 1/1 Running 0 2d
> wikipedia-feed-0-4its2 1/1 Running 0 1d
> wikipedia-parser-0-l0onv 1/1 Running 0 17h
> wikipedia-parser-1-crrxh 1/1 Running 0 17h
> wikipedia-parser-2-1c5nn 1/1 Running 0 17h
> wikipedia-stats-0-3gaiu 1/1 Running 0 16h
> wikipedia-stats-1-j5qlk 1/1 Running 0 16h
> wikipedia-stats-2-2laos 1/1 Running 0 16h
> zookeeper-1-1sb4a 1/1 Running 0 2d
> zookeeper-2-dndk7 1/1 Running 0 2d
> zookeeper-3-46n09 1/1 Running 0 2d
>
>
> Finally, accessing services within the Kubernetes cluster from the outside
> is quite cumbersome unless one uses an external load balancer. This makes
> it difficult to bootstrap a job, as SamzaJob must connect to Zookeeper and
> Kafka to find out the number of partitions on the topics it will subscribe
> to, so it can assign them statically among the number of containers
> requested.
>
> Ideally Samza would operate along the lines of the Kafka high-level
> consumer, which dynamically coordinate to allocate work among members of a
> consumer group. This would do away with the new to execute SamzaJob a
> priori to generate the JobModel to pass to the SamzaContainers. It would
> also allow for dynamically changing the number of containers without having
> the shutdown the job.
>