Tenzing A SQL Implementation On The MapReduce Framework

• Tenzing supports both scalar and table-valued user-defined functions, implemented by embedding a Sawzall interpreter in the Tenzing execution engine. The framework is designed such that other languages can also be easily integrated. In- tegration of Lua and R has been proposed, and work is in progress.（UDF使用swazall实现然后通过内嵌的swazall解释器来运行。同时框架设计上也允许其他语言比如Lua和R来编写UDF）
• Tenzing currently has support for creating func- tions in Sawzall that take tables (vector of tuples) as input and emit tables as output. These are useful for tasks such as normalization of data and doing complex computation involving groups of rows.（所谓的Table Valued Functions就是能够以多个表为输入同时输出多个表，广义一点来看的话就是能够处理vector，这点和scalar对应）

Tenzing supports direct querying of, loading data from, and downloading data into many formats. Various options can be specified to tweak the exact form of input / output.

• For example, for delimited text format, the user can spec- ify the delimiter, encoding, quoting, escaping, headers, etc.
• ColumnIO, a columnar storage system developed by the Dremel team.
• Bigtable, a highly distributed key-value store.
• Protocol buffers stored in compressed record for- mat (RecordIO) and sorted strings format.
• MySQL databases.
• Data embedded in the metadata (useful for testing and small static data sets).

Tenzing is tightly integrated with the Google MapReduce implementation, and we made several enhancements to the MapReduce framework to increase throughput, decrease la-tency and make SQL operators more efficient.（在MapReduce实现上做了一些修改来减少延迟和提高吞吐，以及使得SQL算子操作更有效率）

• Workerpool. One of the key challenges we faced was re-ducing latency from minutes to seconds. It became rapidly clear that in order to do so, we had to implement a solution which did not entail spawning of new binaries for each new Tenzing query（需要将计算的延迟从分钟级别减少到秒级别，在实现上需要使得每次产生新的query不会开辟新的进程）。The MapReduce and Tenzing teams collab-oratively came up with the pool implementation. A typical pool consists of three process groups:
  • The master watcher. The watcher is responsible for receiving a work request and assigning a free master for the task. The watcher also monitors the overall health of the pool such as free resources, number of running queries, etc. There is usually one one watcher process for one instance of the pool.（master watcher接收请求然后将这个任务指派给master，然后这个master去调度这个请求，同时监控总体资源情况）
  • The master pool. This consists of a relatively small number of processes (usually a few dozen). The job of the master is to coordinate the execution of one query. The master receives the task from the watcher and distributes the tasks to the workers, and monitors their progress. Note that once a master receives a task, it takes over ownership of the task, and the death of the watcher process does not impact the query in any way.（针对某个job临时的jobtracker）
  • The worker pool. This contains a set of workers (typi-cally a few thousand processes) which do all the heavy lifting of processing the data. Each worker can work as either a mapper or a reducer or both. Each worker constantly monitors a common area for new tasks and picks up new tasks as they arrive on a FIFO basis. We intend to implement a priority queue so that queries can be tiered by priority.（以FIFO方式完成所有提交的task，可以担任mapper/reducer的角色）
  • Using this approach, we were able to bring down the la-tency of the execution of a Tenzing query itself to around 7 seconds. （使用上面的方式可以将查询延迟降到7s）
  • There are other bottlenecks in the system however, such as computation of map splits, updating the metadata service, committing / rolling back results (which involves file renames), etc. which means the typical latency varies between 10 and 20 seconds currently. （但是还有一些其他瓶颈比如计算map split，更新metadata信息，重命名文件等，这些时间加在一起在10-20s左右变化）
  • We are working on various other enhancements and believe we can cut this time down to less than 5 seconds end-to-end, which is fairly ac-ceptable to the analyst community.
• Streaming & In-memory Chaining. #note: 类似 MapReduce-Online 中提到的做法
  • The original im-plementation of Tenzing serialized all intermediate data to GFS. This led to poor performance for multi-MapReduce queries, such as hash joins and nested sub-selects.（中间物化代价非常高）
  • We im-proved the performance of such queries significantly by im-plementing streaming between MapReduces, i.e. the up-stream and downstream MRs communicate using the net-work and only use GFS for backup.（通过网络传输但是在GFS上进行备份）
  • We subsequently im-proved performance further by using memory chaining, where the reducer of the upstream MR and the mapper of the downstream MR are co-located in the same process.（上下游pipelining）
  • #note: 感觉可能还是spark/shark架构会更加合适，很明显MapReduce方式不太适合DAG计算
• Sort Avoidance. 一些操作不需要做sort可以关闭，这样mapper输出直接就feed给reducer.
  • Certain operators such as hash join and hash aggregation require shuffling, but not sorting. The MapReduce API was enhanced to automatically turn off sorting for these operations.
  • When sorting is turned off, the mapper feeds data to the reducer which directly passes the data to the Reduce() function bypassing the intermediate sorting step. This makes many SQL operators significantly more efficient.
• Block Shuffle. 如果不需要做sort的话那么进行shuffle的时候，那么可以通过将rows聚合成为一个block，以block为单元进行shuffle.
  • Typically, MapReduce uses row based en-coding and decoding during shuffle. This is necessary since in order to sort the data, rows must be processed individu-ally. However, this is inefficient when sorting is not required.
  • We implemented a block-based shuffle mechanism on top of the existing row-based shuffler in MapReduce that combines many small rows into compressed blocks of roughly 1MB in size.
  • By treating the entire block as one row and avoiding reducer side sorting, we were able to avoid some of the over-head associated with row serialization and deserialization in the underlying MapReduce framework code.
  • This lead to 3X faster shuffling of data compared to row based shuffling with sorting.
• Local Execution. 如果处理数据非常小的话那么可以在本地执行而不需要提交到worker pool.
  • The backend can detect the size of the underlying data to be processed. If the size is under a threshold (typically 128 MB), the query is not sent to thepool, but executed directly in the client process. This re-duces the query latency to about 2 seconds.

• Our execution engine has gone through multiple itera- tions to achieve single-node efficiency close to commercial DBMS.
  • The first implementation translated SQL expres- sions to Sawzall code. This code was then compiled using Sawzall’s just-in-time (JIT) compiler. However, this proved to be inefficient because of the serialization and deserializa- tion costs associated with translating to and from Sawzall’s native type system. （第一版通过将SQL表达式编译成为Swazall代码，然后Swazall通过JIT编译器编译然后运行。但是和Sawzall native type system之间转换的序列化和反序列化代价太高）
  • The second and current implementation uses Dremel’s SQL expression evaluation engine, which is based on direct evaluation of parse trees of SQL expressions. While more efficient than the original Sawzall implementa- tion, it was still somewhat slow because of its interpreter-like nature and row based processing.（通过Dremel SQL表达式计算引擎，但是计算引擎实现本质上是解释器，同时处理方式是基于行的）
  • For the third iteration, we did extensive experiments with two major styles of execution: LLVM based native code gen- eration with row major block based intermediate data and column major vector based processing with columnar in- termediate storage.（LLVM生成本地代码运行，并且有两种中间存储格式）