寻网站开发人员合作,定制化网站开发一般多少钱,做网站都得会什么技术,宾馆酒店网站建设方案距离 Google 的上一篇 F1 论文#xff0c;也就是 F1: A Distributed SQL Database That Scales 已经 5 年过去了#xff0c;Google 在今年的 VLDB 上终于发布了 F1 的新版本 F1 Query: Declarative Querying at Scale#xff0c;我们今天就来看一下这篇论文。
2013 年的 F1…距离 Google 的上一篇 F1 论文也就是 F1: A Distributed SQL Database That Scales 已经 5 年过去了Google 在今年的 VLDB 上终于发布了 F1 的新版本 F1 Query: Declarative Querying at Scale我们今天就来看一下这篇论文。
2013 年的 F1 是基于 Spanner主要提供 OLTP 服务而新的 F1 则定位则是大一统旨在处理 OLTP/OLAP/ETL 等多种不同的 workload。但是这篇新的 F1 论文对 OLTP 的讨论则是少之又少据八卦是 Spanner 开始原生支持之前 F1 的部分功能导致 F1 对 OLTP 的领地被吞并了。下面看一下论文的具体内容疏漏之处欢迎指正。
0. 摘要
F1 Query 是一个大一统的 SQL 查询处理平台可以处理存储在 Google 内部不同存储介质Bigtable, Spanner, Google Spreadsheet上面的不同格式文件。简单来说F1 Query 可以同时支持如下功能OLTP 查询低延迟 OLAP 查询ETL 工作流。F1 Query 的特性包括
为不同数据源的数据提供统一视图利用数据中心的资源提供高吞吐和低延迟的查询High ScalabilityExtensibility
1. 背景
在 Google 内部的数据处理和分析的 use case 非常复杂对很多方面都有不同的要求比如数据大小、延迟、数据源以及业务逻辑支持。结果导致了许多数据处理系统都只 focus 在一个点上比如事务式查询、OLAP 查询、ETL 工作流。这些不同的系统往往具有不同的特性不管是使用还是开发上都会有极大的不便利。
F1 Query 就在这个背景下诞生了用论文中的话说就是 we present F1 Query, an SQL query engine that is unique not because of its focus on doing one thing well, but instead because it aims to cover all corners of the requirements space for enterprise data processing and analysis. F1 Query 旨在覆盖数据处理和分析的所有方面。F1 Query 在内部已经应用到了多个产品线比如 Advertising, Shopping, Analytics 和 Payment。
在 F1 Query 的系统设计过程中下面几点考量具有非常关键的作用。
Data Fragmentation: Google 内部的数据由于本身的特性不同会被存储到不同的存储系统中。这样会导致一个应用程序依赖的数据可能横跨多个数据存储系统中甚至以不同的文件格式。对于这个问题F1 Query 对于这些数据提供一个统一的数据视图。Datacenter Architecture: F1 Query 的目标是多数据中心这个和传统的 shared nothing 架构的数据处理系统不同相同。传统的模式为了降低延迟往往需要考虑 locality也就是数据和计算越近越好。由于 Google 内部的网络环境优势locality 的优势显得不是那么重要。所以 F1 Query 更强调计算和存储分离这样计算节点和存储节点的扩展性scalability都会更好。毕竟 Google 内部的系统scalability 才是第一法则。还有一点值得一提的是由于使用了 GFS 的更强版本: Colossue File System磁盘不会成为瓶颈。Scalability: 在 F1 Query 中short query 会在单个节点上执行larger query 会以分布式的模式执行largest query 以批处理 MapReduce 模式执行。对于这些模式F1 Query 可以通过增加运算的并行度来优化。Extensibility: 对于那些无法用 SQL 语义来表达的查询需求F1 通过提供 user-defined functions (UDF)、user-defined aggregate functions (UDAs) 和 table-valued functions (TVF) 来支持。
2. 架构
F1 的架构图如下所示 下面的方框里面是每个 Datacenter 一套。关于各个组件的介绍如下
用户通过 client libary 和 F1 Server 交互F1 Master 负责 query 的状态的运行时监控和其他组件的管理F1 Server 收到用户请求对于 short query 直接单机执行查询对于 larger query 转发到多台 worker 上并行执行查询。最后再汇总结果返回给 client。F1 Worker 负责具体查询执行F1 Server 和 Worker 都是无状态的方便扩展
2.1 query 执行
用户通过 client libary 提交 query 到 F1 Server 上F1 Server 首先解析和分析 SQL然后将涉及到的数据源提取出来如果某些数据源在当前 datacenter 不存在则直接将 query 返回给 client 并告知哪些 F1 Server 距离哪些数据源更近。这里直接将请求返回给业务层由业务层去 retry设计的也是非常的简单。尽管前面说到要将存储和计算分离但是这个地方的设计还是考虑到了 localitydatacenter 级别的 locality毕竟 locality 对查询延迟的影响还是巨大的。
F1 Server 将 query 解析并优化成 DAG然后由执行层来执行具体执行模式interactive 还是 batch由用户指定。原文是: Based on a client- specified execution mode preference, F1 Query executes queries on F1 servers and workers in an interactive mode or in a batch mode.
对于交互式查询模式interactive mode有单节点集中执行模式和多节点分布式执行模式query 优化会根据启发式的算法来决定采用哪种模式。集中式下F1 Server 解析分析 query然后在当前节点上直接执行并接收查询结果。分布式下接收 query 的 F1 Server 充当一个 query coordinator 的角色将 query 拆解并下发给 worker。交互式查询在数据量不太大的情况下往往具有不错的性能和高效的资源利用率。
除了交互式查询还有一种模式是批处理模式batch mode)。批处理模式使用 MapReduce 框架异步提交执行执行相比交互式这种 long-running 方式批处理模式的可靠性reliabitly更高。
2.2 数据源
数据查询支持跨 datacenter。存储计算分离模式使得多数据源的支持更加简单比如 Spanner, Bigtable, CSV, columnar file 等。为了支持多数据源F1 Query 在他们之上抽象出了一层让数据看起来都是存储在关系型表里面。而各个数据源的元数据就存储在 catalog service 里面。
对于没有存储到 catalog service 里面的表数据只要提供一个DEFINE TABLE即可查询。
DEFINE TABLE People(format ‘csv’,path ‘/path/to/peoplefile’,columns ‘name:STRING,DateOfBirth:DATE’);SELECT Name, DateOfBirth FROM PeopleWHERE Name ‘John Doe’;论文中没有提到的是单看这个 DEFINE TABLE 可以表现力不够所说这些信息并不足以表现出数据的行为
是否支持 partition?是否支持 逻辑下推是否支持索引是否支持多种 扫描模式对于新数据源的支持可以通过 Table-Valued Function (TVF) 的方式来支持。
2.3 Data Sink
query 的结果可以直接返回给 client也可以插入到另外一个表里面。
2.4 SQL
SQL 2011。之所以是 2011 是因为其他老的系统使用的是 2011。
3. 交互式查询
交互式查询模式是默认的查询模式。如前所述交互式查询有集中式和分布式具体使用哪种由优化器分析 client 的 query 然后决定。
3.1 Single Threaded Execution Kernel
集中式的查询如下图所示是一种 pull-based 的单线程执行方式。 3.2 Distributed Execution
如前所述由优化器分析完 query 决定是否采用分布式模式。在分布式这种模式下接收到 query 的 F1 Server 充当一个 coordinator 的角色将执行 plan 推给 worker。worker 是多线程的可以并发执行单个 query 的无依赖的 fragment。Fragment 是执行计划切分出来的执行计划片段非常像 MR 或者 Spark 中的 stage。Fragment 之间通过 Exchange Operator (数据重分布) 连接。
Fragment 的切分过程如下优化器使用一种基于数据分布依赖的 bottom-up 策略。具体来说每个算子对于输入数据的分布都有要求比如 hash 或者依赖其他字段的分布。典型的例子有 group by key 和 hash join。如果当前的数据分布满足前后两个算子的要求则两个算子就被放到一个 Fragment 里面否则就被分到两个 Fragment 里面然后通过 Exchange Operator 来连接。
下一步就是计算每个 Fragment 的并行度Fragment 之间并行度互相独立。叶子节点的 Fragment 的底层 table scan 决定最初的并行度然后上层通过 width calculator 逐步计算。比如 hash-join 的底层两个 Fragment 分别是 100-worker 和 50-worker则 hash-join 这个 Fragment 会使用 100-worker 的并行度。下面是一个具体的例子。
SELECT Clicks.Region, COUNT(*) ClickCountFROM Ads JOIN Clicks USING (AdId)WHERE Ads.StartDate ‘2018-05-14’ ANDClicks.OS ‘Chrome OS’GROUP BY Clicks.RegionORDER BY ClickCount DESC;上面 SQL 对应的 Fragment 和一种可能 worker 并行度如下图所示 3.3 Partitioning Strategy
数据重分布也就是 Fragment 之间的 Exchange Operator对于每条数据数据发送者通过分区函数来计算数据的目的分区数每个分区数对应一个 worker。Exchange Operator 通过 RPC 调用扩展可以支持到每个 Fragment 千级的 partion 并发。要求再高就需要使用 batch mode。
查询优化器将 scan 操作作为执行计划的叶子节点和 N 个 worker 节点并发。为了并发执行 scan 操作数据必须要被并发分布然后由所有 worker 一起产生输出结果。有时候数据的 partition 会超过 N而 scan 并发度为 N多余的 partition 就交由空闲的 worker 去处理这样可以避免数据倾斜。
3.4 Performance Considerations
F1 Query 的主要性能问题在于数据倾斜和访问模式不佳。Hash join 对于 hot key 尤为敏感。当 hot key 被 worker 载入到内存的时候可能会因为数据量太大而写入磁盘从而导致性能下降。
论文中举了一个 lookup join 的例子这里不打算详述了。
对于这种数据倾斜的问题F1 Query 的解决方案是 Dynamic Key Range但是论文中对其描述还是不够详细。
F1 Query 对于交互式查询采用存内存计算而且没有 check point。因为是内存计算所以速度非常的快但是由于没有 checkpoint 等 failover 的机制只能依赖于业务层的重试。
4. 批处理
像 ETL都是通过 Batch Mode 来处理的。Google 以前都是通过 MapReduce 或者 FlumeJava 来开发的开发成本一般比较高。相比 SQL 这种方式不能有效复用 SQL 优化所以 F1 Query 选择使用 SQL 来做。
如前所述交互式查询不适合处理 worker failure而 batch mode也就是批处理这种模式特别适合处理 failover每一个 stage 结果落盘。批处理模式复用交互式 SQL query 的一些特性比如 query 优化执行计划生成。交互式模式和批处理模式的核心区别在于调度方式不同交互式模式是同步的而批处理模式是异步的。
4.1 Batch Execution Framework
批处理使用的框架是 MapReduceFragment 被抽象成 MapReduce 中的 stagestage 的输出结果被存储到 Colossus file system (GFS 二代)。
在 Fragment 映射有一点值得注意的是严格来说Fragment 的 DAG 映射到 mr 是 map-reduce-reduce对这种模式做一个简单的变通变成map-reduce-mapidentity-reduce如下图 关于 MapReduce 的更详细信息可以参考 Google 03 年那篇论文。
4.2 Batch Service Framework
Framework 会对 batch mode query 的执行进行编排。具体包括query 注册query 分发调度已经监控 mr 作业的执行。当 F1 Server 接收到一个 batch mode query它会先生成执行计划并将 query 注册到 Query Registry全局唯一的 Spanner db用来 track batch mode query。Query Distributor 然后将 query 分发给 datacenter。Query Scheduler 会定期从 Registry 拿到 query然后生成执行计划并交给 Query Executor 来处理。
Service Framework 的健壮性非常好Query Distributor 是选主master-elect模式Query Scheduler 在每个 datacenter 有多个。query 的所有执行状态都是保存在 Query Registry这就保证其他的组件是无状态的。容错处理MapReduce 的 stage 会被重试如果 datacenter 出问题query 会被分配到新的 datacenter 上重新执行。
5. 优化器
SQL 优化器类似 Spark Catalyst架构如下图不细说了。
6. EXTENSIBILITY
对于很多复杂业务逻辑无法用 SQL 来描述F1 针对这种情况提供了一种用户自定义函数的方法包括 UDF (user-define functions)UDA (aggrega- tion functions) 和 TVF (table-valued functions)。对于简单的UDF需求通常直接以SQL或者LUA的形式作为query的一部分对于更复杂或者性能要求高的UDF需求则可以用其它高级语言以UDF Server的形式实现。
UDF Server 和 F1 Query 是 RPC 调用关系有 client 单独部署在同一个 datacenter。udf server 完全有 client 来控制无状态基本可以无限扩展。
6.1 Scalar Functions
UDF 并不是新的概念UDF Server 这种部署方式看上去还算新颖一点。但是 UDF Server 这种单独部署模式一个可能的问题是延迟问题这里通过批量流水线的方式来减少延迟。下面是 UDF 的一个例子。
local function string2unixtime(value)local y,m,d match((%d)%-(%d)%-(%d))return os.time({yeary, monthm, dayd})
end6.2 Aggregate Functions
UDA 是对多行输入产生一个单一的输出要实现 UDA用户需要实现算子 Initialize, Accumulate, and Finalize。另外如要要对多个 UDA 的子聚合结果进行再聚合用户可以实现 Reaccumulate。
6.3 Table-Valued Functions
TVF 的输入是一个 table输出是另外一个 table。这种在机器学习的模型训练场景下比较有用。下面是论文中的具体的一个例子EventsFromPastDays 就是一个 TVF。
SELECT * FROM EventsFromPastDays(3, TABLE Clicks);
当然 TVF 也支持用 SQL 来描述如下。
CREATE TABLE FUNCTION EventsFromPastDays(num_days INT64, events ANY TABLE) ASSELECT * FROM eventsWHERE date DATE_SUB(CURRENT_DATE(),INTERVAL num_days DAY);
7. Production Metric
下面是 F1 Query 在 Production 环境下的几个 metrics。 8. 总结
回过头来看 F1 Query 最新的这篇论文给人最大的启发就是大一统的思想这个很有可能是行业发展趋势。回想一下 MapReduce 论文由 Google 于 2003 年发表开源实现 Hadoop 于 2005 问世。不妨期待了一下未来的 3 到 5 年的 F1 Query 的开源产品。
原文链接 本文为阿里云原创内容未经允许不得转载。
