Pinot on Yarn

[subprotocol]

Placeholder to discuss functionality that would enable Pinot to run on Yarn.

Some initial thoughts- https://groups.google.com/forum/#!topic/pinot_users/RL7QrAdAIf8

I am interested in writing a small yarn app that would deploy Pinot into Yarn. This would make it easier for my group to deploy pinot into clustered environments with a single command. I wanted to run this by folks here to gauge the level of interest of having this be a potential contribution.

As a starting point, this is what I envision-

1. a yarn deploy class, and possibly a script wrapper (pinot-yarn-deploy.sh)

PinotYarnDeploy (and a few additional ancillary classes) would be created and placed inside com.linkedin.pinot.hadoop.yarn from the pinot-hadoop module (or possibly as its own module if that would make more sense: pinot-yarn?)

PinotYarnDeploy would leverage args4j to accept yarn and pinot properties such as numControllers, numBrokers, numServers, and settings for each container type (cpus, memory, container overhead, etc..).

2. a basic yarn application master with a web ui that would:
   a. show where things are running
   b. allow one to add/remove: controllers, brokers, servers from the yarn app master ui

Running in yarn all the storage modes should be able to be supported.

• hdfs backed- containers would copy segments out of hdfs, which is extra copying and storage but should be fine

• nfs backed (we'd probably use the hdfs nfs gateway as that allows storage in hdfs, supports mmapping files directly out of hdfs).

Beyond that I would be interested in adding the following as well:

• allow services to bind to random ports if passed 0 as the port number

• pinto-admin.sh commands that take -controllerHost -controllerPort would optionally be able to take -zkAddress, which would choose a controller at random, and then connect to it using the published host and port that sit in zookeeper (a.k.a- this just makes it easy to work with random ports)

Any input is appreciated and welcome, thank you!

[subprotocol]

As I currently see it, this would encompass the following additions-

• add yarn deploy related classes, ie- PinotYarnDeploy, PinotYarnAppMaster, ... also possibly a script wrapper (pinot-yarn-deploy.sh)
• provide yarn application master that manages container lifecycle, and has a ui component that shows where services are running, ports, etc, ..
• allow services to bind to random ports if passed 0 as the port number
• change pinot-admin.sh for commands that require a controller host and port, to optionally take -zkAddress. This would discover and choose a random controller to send the command to

[kishoreg]

I updated the issue with the content of the email. Hope that's ok.

We are definitely interested in this initiative. We were thinking of building something like this for other provisioning systems such as AWS, Azure, Mesos, Kubernetes, Rain (internal to LinkedIn). It will be great if we can build right abstractions so that most of the code can be re-used across different systems.

Another way to look at this problem would be from Helix point of view. It would be cool for Apache Helix project add the support for deploying a Helix based system on any provisioning system.

Some pointers -

• @jfim has added the support for Pinot on Kubernetes.
• Gobblin on Yarn
• Redis auto-scaling with Helix

We can definitely start with a PoC for Yarn and extend it to other systems as the second step. Can you come up an architecture on how this would look when deployed on YARN.

[subprotocol]

Thank you for the resources. I absolutely agree on nailing down the right abstractions. I am thinking more through the ways this could come together. This is what I have pictured currently-

PinotSubmit: Determines and calls the launcher to submit Pinot into the cluster (YarnLauncher, MesosLauncher, ..)

ClusteredService: Interface for abstractions to the cluster backend, similar to what Gobblin did (YarnService, MesosService, .. would implement this)

I don't think there is too much commonality between launchers (they are just full of backend specific boilderplate). In Spark for example, their launcher (https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala) determines the backend and just forwards the arguments to the appropriate launcher class (ie- org.apache.spark.deploy.yarn.YarnClusterApplication, org.apache.spark.deploy.k8s.submit.KubernetesClientApplication, ...). Those launcher classes are optionally available depending on how spark was compiled. I imagine it would make sense to do something similar for Pinot. And this should allow compatibility with something internal like Rain; just add a jar onto the class path that defines the RainLauncher and implements ClusteredService.

ClusteredService would abstract acquiring and releasing resources. What do we want this to mean exactly, and who should call it? Would it make the most sense to just listen to when a tenant is added/updated/removed, and allocate/deallocate containers based off that?

If that's the case ClusteredService could just be an abstract class that knows how to listen to tenant changes. When a tenant change occurs, it simply passes it on to the implementing backend, such as YarnService.

YarnService would then just need to implement something like: tenantAdd, tenantUpdate, tenantDelete. When a tenant is updated from 5 to 8 real-time servers, it would know to allocate 3 containers. When updated to from from 8 to 6, YarnService would release 2 of that tenants containers (or listen for when they transition back to free/un-tagged).

As for how this would look on Yarn-

1. pinot-submit.sh -cluster=yarn -zkAddress=zookeeper -controllers 3 <other args..>
2. YarnLauncher would get called which creates the cluster in helix if it doesn't exist and submit app into yarn
3. YarnService (an instance of ClusteredService) would start up inside the cluster, register as the AppMaster, and then create 3 pinot controller containers
4. User adds a tenant
5. ClusteredService sees that a tenant was added and calls tenantAdd
6. YarnService implements tenantAdd would create the containers
   ...

<other args> would specify things like how to size the container types (real-time server vs. historical server, broker, controllers) with settings such as heap size, container overhead, other jvm settings, ...

Let me know if I'm on the right track. Thanks again!

[kishoreg]

This looks good to me. In step 4, instead of adding tenant, we can model it as add node or add instance which can take instanceType as an argument. InstanceType can be Server,Broker, Minion etc.

This should have set up the cluster. After this users can invoke addTable etc which will take tableConfig as an argument. TableConfig has the tenant info and we automatically assign instances to a tenant. We have all this available as of today.

Thoughts?

[subprotocol]

Ahhh! Ok yes, I think I follow what you mean. I believe this should work nicely. I will spend some cycles digging in a bit more in the morning.

One question I have- If instanceType is SERVER, my understanding is that the server type (realtime/offline) won't be determined until the instance is created/assigned to a server. If we want realtime servers to have say--a larger heap than offline servers--is there a way to designate the type that a server can be in advance?

[kishoreg]

Good point. We have two options we can overload the instanceType to SERVER_OFFLINE, SERVER_REALTIME, BROKER, MINION and/or take additional arguments such as tags and this can be any arbitrary tags or tenant name.

addInstance
  -- instanceType(s) <SERVER_REALTIME|SERVER_OFFLINE|BROKER|MINION> //required param
  -- tenantName(s) <list of tenant that this instance can be belong to>//optional set to "default" by default :-)
  -- tags  <additional tags to have fine grained control on segment/table assignments>

[subprotocol]

@kishoreg I have a very basic prototype that is able to run Pinot entirely in YARN. Currently the number of controllers/brokers/servers needs to be specified up front, but they can be added/removed during runtime.

You had mentioned the Minion instance type- I'm trying to figure out where exactly this ties in. Am I correct in thinking that the Minion is work still in progress, but will eventually have a PinotAdministrator StartMinion command?

I'm now looking closer at modeling this all around addInstance per your input.

Originally I was thinking the cluster manager backend would listen as a participant when an instance is needed. But Pinot can't assign instances that don't already exist. Given that, would addInstance be a new command within PinotAdministrator that would communicate with the cluster manager backend to create the resources?

Example workflow would be something like-

1. submit application into yarn
2. run AddInstance commands-
   a. AddInstance -instanceType BROKER -tenantName MyTenant -numInstances 2
   b. AddInstance -instanceType SERVER_REALTIME -tenantName MyTenant -numInstances 3
   c. AddInstance -instanceType SERVER_OFFLINE -tenantName MyTenant -numInstances 5
   d. AddInstance -instanceType MINION -tenantName MyTenant -numInstances 1

After running that you'd have a Pinot cluster with 3 controllers, 2 brokers, 3 offline servers, 5 online servers, and 1 minion. Is my thinking aligned with what you are envisioning?

Thanks

[xiangfu0]

Closing — no longer relevant. Kubernetes is now the standard deployment platform for Pinot. Yarn-based deployment was never implemented and is no longer a priority.