Pig on Tez: Low Latency Data Processing with Big Data

Editor's Notes

• #4: First, let me introduce Pig on Tez project team. This team is kind of special compare to other projects. There is no single company driving the development, instead, we have a virtual team which consists of Rohine, Olga from Yahoo!, Cheolsoo from Netflix, Mark and Alex from Linkedin and me from Hortonworks. We work independently but in a very coordinated manner. We have weekly standup, we have sprint, and we use Apache to cooperate. And this model turn out works very well. Actually Cheolsoo and Mark gives a talk in last year’s ApacheCon to talk about this model.
• #5: Let me spend one slide to talk about what is Pig. Pig is a data processing language. It is SQL like but there are some differences. Pig is a procedural language, you process the data step by step. Here is one example of Pig script. In this example….. Unlike SQL, you don’t have to scramble everything into a single SQL statement, which is more natural and intuitive. Pig can process a hdfs file directly. Which means you don’t have to create a table first. You don’t even need schema. Pig is feature rich. It has the features SQL doesn’t have, such as …. Pig is also turing complete. You can write Pig Macro, you can embed Pig inside Python script so you can do things like loop, branching, which you cannot do in SQL.
• #6: We typically see two major group of people who are using Pig. ETL people use Pig because Pig syntax is very similar to ETL tools. Pig operator such as filter, foreach, group, join, are also exist in ETL tools. Another group of people who are using Pig are data scientists. They need richer features which SQL cannot provide, such as nested operators, they like to use python to write a UDF, they want to do looping in their script. We see a lot of companies are using Pig. At Yahoo! …. Other heavy users include…..
• #7: Pig traditionally build on top of Mapreduce. Mapreduce is good, it is stable and scalable. Mapreduce provide a simple API, but in the mean time, very restrictive. Things works very well initially. However, when more and more people migrate their workload to Pig, more and more people depends on Pig to do their daily job, they request more. They want Pig to be fast. However, with the restriction of Mapreduce, Pig cannot process the data as fast as it can. So we are looking for a new engine. This new engine must be general DAG based, that’s what a traditional database engine based on, so that we can do everything a traditional database engine can do. We need a powerful API to deal with the complex requirement Pig needs. In the meantime, we don’t want to change things completely. We want to continue to use Hadoop cluster, we like the stability and scalability of Hadoop, we want to continue to use the enterprise features of Hadoop, such as multi-tenancy, security, encryption, rolling upgrade. After we evaluation different kinds of DAG execution engine, we feel Tez fit all out needs the best. Tez is a low level DAG execution engine. The API is much complex than MapReduce so it is not end user facing. However, for things like Pig, Hive, Cascading, that’s not a problem. These tools can hide the engine complexity internally and not expose to the user. Tez offers a rich API. We can define vertex, edge with semantics, plugin to customize the DAG behavior. Tez is fast, whether you have a lot of memory or not. If you have memory, Tez will make use of it. If not, Tez still perform much faster than Mapreduce. Tez is resource efficient. Compare to Mapreduce, it needs less processing node. Tez is build on top of Yarn, so it inherit a lot of features of Yarn natively, it is multi-tenancy enabled, it is scalable, it is relatively stable as a new engine since it leverage a lot of existing Mapreduce code, you can use Kerboros for security, etc. Last but not the least, the Tez team is awesome. Thanks Bikas, Sidd, Hitesh, they provide excellent community support, if you hit issues, you can count on them to fix it quickly.
• #11: Let’s take a deeper look of the Tez DAG to see what exactly happens. Vertex 1 load the data from hdfs only once. Just like mapreduce, it will partition the data and shuffle to different downstream vertexes. Notice Vertex 1 has two output, so the data are partitioned and shuffled on different key independently to vertex 2 and vertex 3. At vertex 2 and vertex 3, data are merged and sorted. After processing the data, vertex 2 and vertex 3 will partition and shuffle the data on the same key to vertex 4, which will perform a join. We can see the processing of Tez is very similar to Mapreduce, except that we don’t need to store the intermediate file into hdfs. All the data movement are necessary since we need to do a shuffle to perform group and join. So this DAG is very optimal to process the Pig script.
• #13: Pig uses a lot of features and customizations of Tez. We need to customize the vertex, we need to customize input and output between vertices. We need to use vertex manager to tune the parallelism of vertices. Here I will give you some examples. Vertex needs to be customized, that’s obvious. Vertex is the place Pig run data pipeline, so we need to customize the vertex to process data properly. We need to customize edge as well. Unlike MR, which blindly partition, shuffle and sort data between map and reduce, we can do more flexible customization in Tez. In many cases, sorting is not necessary. For example, union only need to combine two inputs without sorting, sample file in order by and right table in replicated join needs to be broadcast without sorting. In MR, we only do scatter-gather between map and reduce. If we want to do a broadcast, such as distribute a scalar, we will have to use a hack way. The most common hack is to put the file on hdfs, and every task read from hdfs. This is very problematic since with thousands read from hdfs, we can easily take namenode down. A better hack is to use distributed cache, while this is much better than hdfs approach, we still need to hit hdfs multiple times since every nodemanager need to read from hdfs and localize the distributed cache. In tez, we can use a standard broadcast edge to do that. We will read hdfs only once and broadcast the file over the network. Another edge type is 1-1 edge, in that one task will and will only process the output of a particular upstream task. Pig use 1-1 edge in order by and skewed join. In such case, tez will schedule the two task on the same node manager thus completely avoid network traffic. Besides vertices and edges, we can further customize the DAG behavior through a plugin called VertexManager. When the DAG progress, VertexManager will receive notification of the DAG and adjust vertex and edge accordingly. Currently Pig use VertexManager to adjust the vertex parallelism dynamically.
• #14: For every Mapreduce job, we will need a AM. The job of AM is to launch map tasks and reduce tasks. After the mapreduce job complete, we kill the AM. The next mapreduce job will launch another AM. A Pig script usually contains dozens of MR jobs, so that’s a significant delay and waste of resources since AM is only doing admin work not the real work we need MR to do. This often leads to resource congestion. In one extreme case, we see hundreds of AM is running on a cluster but one real work can be done since there is no container available to do the real work, thus no one can make progress. Tez solve this problem completely. For a Tez DAG which usually represents the workload of a dozens of MR jobs, require only one AM. Further, even if the DAG finishes, Tez does not kill the AM, instead, the same AM can be reused to launch another DAG, which is called session reuse. This benefit Pig a lot. Typically, we will compile a Pig script into a single DAG, so that only one AM is required. There are couple of cases which Pig need to submit multiple DAGs, such as multiple exec statement, multiple DAG launched by a single grunt session, multiple DAGs generated by a loop inside a python scripts. In that, Pig maintain a session pool and submit the DAG to existing AM whenever possible.
• #15: Container reuse is even more useful. Container is the place where Pig run the data pipelines. In MR, once the task finishes, we kill container jvm. A large Mapreduce job may consists of several thousand tasks, which means we need to kill and restart jvm thousands of times. Further, in some vertex, we need to spend time to initialize the resource, for example, in replicated join, we need to load the right side relation and put it in memory, then we will need to do that thousands of times as well. In a congested cluster, request a new container means wait in line, and compete resource with other jobs, and we need to to thousands of times. That’s a significant delay of processing. In Tez, we don’t kill the container immediately after the task finishes. Instead, we try to reuse the container whenever possible. The first task can cache any memory object into a key-value store called ObjectCache. The next task will retrieve the object by name from the ObjectCache. Pig use ObjectCache to store the right relation of the right table in replicated join, sample file in order by and skewed join. With container reuse and ObjectCaching, we minimize the cost of kill and restart jvm, initialize the resource, and compete the resource in a busy cluster. We see significant speedup for replicated join, and we see order of magnitude speedup for smaller job. One thing to note, however, custom Pig LoadFunc/StoreFunc/UDFs might need to be hardened. Since Tez will reuse the same jvm, static variable needs to be reinitalized when vertex started, and some memory leak which is not obvious in MR may cause issues in Tez.
• #16: Another optimization is vertex group. Look at the example script, if we do it in mapreduce, we will need 2 mapreduce job to process it. The first mapreduce job will process 2 group by simultaneously. Thanks to multiquery optimization, otherwise we will need 3 mapreduce jobs. The second mapreduce job only do a union. Its job is very simple, just combining two outputs together. In Tez, we introduce a concept called vertex group. It is a virtual vertex which moves outputs of two group vertex into the same folder, thus avoid a real Tez task.
• #17: In Pig, user can set parallelism manually by parallel statement or define a global default_parallel in Pig script. However, set parallelism manually is hard. User usually don’t have idea how to set parallelism properly. In MR, we use automatic parallelism if user leave the parallelism blank and we continue to do that in Tez and we do it better. There are multiple layers of parallelism adjustment. At compile time, we estimate the parallelism of each vertex based on the DAG input and the data pipeline inside each vertex. This is very rough however, unlike Hive, which is able to collect data stats and save into metastore, Pig don’t have the stats of the input data. When the DAG progress, we will find we underestimate or overestimate the parallelism originally. VertexManager provide an opportunity to monitor the DAG progress and adjust the parallelism dynamically.
• #18: When a vertex is about to start, VertexManager will estimate its input data, and adjust the parallelism of the vertex. This is called dynamic parallelism. The most common VertexManager is already implemented in Tez, which is the ShuffleVertexManager. It is able to handle typical Scatter-gather edge. There are other VertexManagers available in Tez. And of cause, you can implement customized VertexManager, and Pig did that for order by and skewed join.
• #19: Let’s take a deeper look into ShuffleVertexManager. It is widely used in Pig to perform most operation needs shuffle, such as group, hash join. In this scenario, we initially set the parallelism 4 for the join vertex. When the JOIN vertex is about to start, we find the actual data coming out of A and B is lesser than estimated. We will only need parallelism 2 according to our policy. ShuffleVertexManager will get this notification, and decide to change the parallelism. So we eliminated task #3 and #4. Notice vertex A and vertex B is already started, so the output data is already partitioned into 4 partitions. Tez will do something very sophisticated, it will reroute the partition originally going to task 3 and task 4 to task 1 and task 2. Thus, parallelism of JOIN vertex is changed to 2 at runtime. Notice the way ShuffleVertexManager estimate the input data size of JOIN vertex. It wait for a certain percentage of input task finishes, then estimate the input size based on the finished tasks. If the input tasks are highly skewed, this is problematic. Also, ShuffleVertexManager only decrease the parallelism but not increase. Decrease the parallelism only need to reroute the eliminated partition to existing task. On the other hand, increase the parallelism require to repartiton the existing partitions, which is harder to do.
• #20: Order by, on the other hand, estimate the parallelism based on the samples collected. In order by, we will first sample the input data to decide how many tasks to be used in sorting, and the key range for every task. This number will send to the VertexManager of the sorting vertex. Since the upstream node is not started yet, which means input data is not partitioned, we can freely adjust the parallelism of the sorting vertex, either decrease parallelism, or increase parallelism. The upstream vertex will then partition the data with the right parallelism. And because the data size is estimated based on a complete random sample of the input data, it is much more accurate.
• #21: When we are discussing about ShuffleVertexManager, we already covered limitation of it. We can only decrease parallelism but not increase it. Also, there is overhead involved when we reroute the partition. So it is better to get things right before the upstream vertex start. This is why we need another layer of parallelism adjustment, and we call it Pig grace parallelism. The idea is when the DAG progress, we adjust the parallelism of the downstream vertex accordingly with the better estimation. Even when ShuffleVertexManager cannot bring everything right, it will not off too much.