基于spark服务行业的用户行为分析（基于spark服务行业的用户行为分析报告）

KTV免费预定 2022年12月06日 15:11:34 1

默认

本文目录一览：

1、用户画像-轻量级基于spark人群圈选
2、技术解析Transwarp Inceptor是怎样炼成的
3、大数据技术中，关于用户行为分析方面的有哪些技术
4、可以在Windows环境下用spark分析网络用户行为吗

用户画像-轻量级基于spark人群圈选

人群圈选，也叫人群定向。在业界有中广泛的业务场景。

业界比较优秀的方案有百度基于doris来实现海量用户的圈选，可以实现千万级人群秒级圈选。但是这种方案比较复杂：

1、需要数仓同学加工二值化tag，构建用户二值化tag到用户bitmap集合的倒排索引

2、需要通过构建哈希分桶列，解决超大bitmap基数交并集问题

3、需要数仓同学构建全局字典，防止不连续id带来的roaring bitmap性能问题

4、需要通过to_bitmap函数解决动态标签与静态标签的组合圈选问题

5、需要团队有doris的运维能力

本博文介绍一种基于spark的轻量级圈选，可以实现千万级人群分钟级（5分钟以内）圈选。

业界还有基于ES、MR、离线+服务bitmap等方案（我们选型时，doris还不火，故不在此列）。

ES：大人群的滚动导出会给集群带来很大压力的；另外就是如果构建一个大而全画像宽表的话，这个表会很稀疏，所以我们会按业务主题构建多个es表，这样就会涉及到多个表之间join的问题，而es是不支持join语义的。

MR：老方案了，性能不大行

es+bitmap方案：这个方案扩展起来有点doris的味道。但是自己开发的话，其实比较复杂，而且没有居多性能优化手段的话，性能没那么理想，大都是比不上spark的。

因此，最后我们选择了spark做为人群圈选的方案。

我们构建了用户画像相关的一站式平台，前端平台包括人群管理、标签管理、画像分析等相关功能。

中间层服务对应包含人群管理中心、人群任务调度、人群解析引擎（负责前端json到Spark SQL的转化）、标签管理中心的模块；当然也包含权限、安全、流控、人群监控等辅助模块。

存储层，我们主要基于spark进行人群定向，依托Hbase/Redis做用户单到点查询，用Mysql做配置信息存储。

我们选取两个主题的画像进行交集圈选，圈选结果千万级，测试结果如下：

基于spark服务行业的用户行为分析（基于spark服务行业的用户行为分析报告）

技术解析Transwarp Inceptor是怎样炼成的

当前Hadoop技术蓬勃发展，用于解决大数据的分析难题的技术平台开始涌现。Spark凭借性能强劲、高度容错、调度灵活等技术优势已渐渐成为主流技术，业界大部分厂商都提供了基于Spark的技术方案和产品。根据Databricks的统计，目前有11个商业的Spark版本。

在使用Spark作出计算平台的解决方案中，有两种主流编程模型，一类是基于SparkAPI或者衍生出来的语言，另一种是基于SQL语言。SQL作为数据库领域的事实标准语言，相比较用API(如MapReduceAPI，SparkAPI等)来构建大数据分析的解决方案有着先天的优势：一是产业链完善，各种报表工具、ETL工具等可以很好的对接;二是用SQL开发有更低的技术门槛;三是能够降低原有系统的迁移成本等。因此，SQL语言也渐渐成为大数据分析的主流技术标准。本文将深入解析Inceptor的架构、编程模型和编译优化技术，并提供基准测试在多平台上的性能对比数据。

1.Inceptor架构

TranswarpInceptor是基于Spark的分析引擎，如图1所示，从下往上有三层架构：最下面是存储层，包含分布式内存列式存储(TranswarpHolodesk)，可建在内存或者SSD上;中间层是Spark计算引擎层，星环做了大量的改进保证引擎有超强的性能和高度的健壮性;最上层包括一个完整的SQL99和PL/SQL编译器、统计算法库和机器学习算法库，提供完整的R语言访问接口。

TranswarpInceptor可以分析存储在HDFS、HBase或者TranswarpHolodesk分布式缓存中的数据，可以处理的数据量从GB到数十TB，即使数据源或者中间结果的大小远大于内存容量也可高效处理。另外TranswarpInceptor通过改进Spark和YARN的组合，提高了Spark的可管理性。同时星环不仅仅是将Spark作为一个缺省计算引擎，也重写了SQL编译器，提供更加完整的SQL支持。

同时，TranswarpInceptor还通过改进Spark使之更好地与HBase融合，可以为HBase提供完整的SQL支持，包括批量SQL统计、OLAP分析以及高并发低延时的SQL查询能力，使得HBase的应用可以从简单的在线查询应用扩展到复杂分析和在线应用结合的混合应用中，大大拓展了HBase的应用范围。

2.编程模型

TranswarpInceptor提供两种编程模型：一是基于SQL的编程模型，用于常规的数据分析、数据仓库类应用市场;二是基于数据挖掘编程模型，可以利用R语言或者SparkMLlib来做一些深度学习、数据挖掘等业务模型。

2.1SQL模型

TranswarpInceptor实现了自己的SQL解析执行引擎，可以兼容SQL99和HiveQL，自动识别语法，因此可以兼容现有的基于Hive开发的应用。由于TranswarpInceptor完整支持标准的SQL 99标准，传统数据库上运行的业务可以非常方便的迁移到Transwarp Inceptor系统上。此外Transwarp Inceptor支持PL/SQL扩展，传统数据仓库的基于PL/SQL存储过程的应用(如ETL工具)可以非常方便的在Inceptor上并发执行。另外Transwarp Inceptor支持部分SQL 2003标准，如窗口统计功能、安全审计功能等，并对多个行业开发了专门的函数库，因此可以满足多个行业的特性需求。

2.2数据挖掘计算模型

TranswarpInceptor实现了机器学习算法库与统计算法库，支持常用机器学习算法并行化与统计算法并行化，并利用Spark在迭代计算和内存计算上的优势，将并行的机器学习算法与统计算法运行在Spark上。例如：机器学习算法库有包括逻辑回归、朴素贝叶斯、支持向量机、聚类、线性回归、关联挖掘、推荐算法等，统计算法库包括均值、方差、中位数、直方图、箱线图等。TranswarpInceptor可以支持用R语言或者SparkAPI在平台上搭建多种分析型应用，例如用户行为分析、精准营销、对用户贴标签、进行分类。

3.SQL编译与优化

TranswarpInceptor研发了一套完整的SQL编译器，包括HiveQL解析器、SQL标准解析器和PL/SQL解析器，将不同的SQL语言解析成中间级表示语言，然后经过优化器转换成物理执行计划。SQL语言解析后经过逻辑优化器生成中间级表示语言，而中间表示语言再经过物理优化器生成最终的物理执行计划。从架构上分，逻辑优化器和物理优化器都包含基于规则的优化模块和基于成本的优化模块。

为了和Hadoop生态更好的兼容，Inceptor为一个SQL查询生成MapReduce上的执行计划和Spark上的执行计划，并且可以通过一个SET命令在两种执行引擎之间切换。

3.1SQL编译与解析

TranswarpInceptor的SQL编译器会根据输入的SQL查询的类型来自动选择不同的解析器，如PL/SQL存储过程会自动进入PL/SQL解析器并生成一个SparkRDD的DAG从而在Spark平台上并行计算，标准SQL查询会进入SQL标准解析器生成Spark或MapReduce执行计划。由于HiveQL和标准的SQL有所出入，为了兼容HiveQL，Transwarp Inceptor保留了HiveQL解析器，并可以对非标准SQL的Hive查询生成Spark或者Map Reduce执行计划。

3.1.1SQL标准解析器

TranswarpInceptor构建了自主研发的SQL标准解析器，用于解析SQL99 SQL 2003查询并生成Spark和Map Reduce的执行计划。词法和语法分析层基于Antlr语法来构建词法范式，通过Antlr来生成抽象语义树，并会通过一些上下文的语义来消除冲突并生成正确的抽象语义树。语义分析层解析上层生成的抽象语义树，根据上下文来生成逻辑执行计划并传递给优化器。首先Transwarp Inceptor会将SQL解析成TABLE SCAN、SELECT、FILTER、JOIN、UNION、ORDER BY、GROUP BY等主要的逻辑块，接着会根据一些Meta信息进一步细化各个逻辑块的执行计划。如TABLE SCAN会分成块读取、块过滤、行级别过滤、序列化等多个执行计划。

3.1.2PL/SQL解析器

PL/SQL是Oracle对SQL语言的模块化扩展，已经在很多行业中有大规模的应用，是数据仓库领域的重要编程语言。

为了让存储过程在Spark上有较好的性能，PL/SQL解析器会根据存储过程中的上下文关系来生成SQLDAG，然后对各SQL的执行计划生成的RDD进行二次编译，通过物理优化器将一些没有依赖关系的RDD进行合并从而生成一个最终的RDDDAG。因此，一个存储过程被解析成一个大的DAG，从而stage之间可以大量并发执行，避免了多次执行SQL的启动开销并保证了系统的并发性能。

解析并生成SQL级别的执行计划

3.2SQL优化器

TranswarpInceptor使用Spark作为默认计算引擎，并且开发了完善的SQL优化器，因此在大量的客户案例性能测试中，TranswarpInceptor的性能领先MapReduce 10-100倍，并超越部分开源MPP数据库。SQL优化器对平台性能的提升居功至伟。

3.2.1基于规则的优化器(RuleBasedOptimizer)

目前为止，TranswarpInceptor共实现了一百多个优化规则，并且在持续的添加新的规则。按照功能划分，这些规则主要分布在如下几个模块：

文件读取时过滤

在文件读取时过滤数据能够最大化的减少参与计算的数据量从而最为有效的提高性能，因此TranswarpInceptor提供了多个规则用于生成表的过滤条件。对于一些SQL中的显示条件，TranswarpInceptor会尽量将过滤前推到读取表中;而对于一些隐式的过滤条件，如可以根据joinkey生成的过滤规则，Inceptor会根据语义保证正确性的前提下进行规则生成。

过滤条件前置

TranswarpInceptor能够从复杂的组合过滤条件中筛选出针对特定表的过滤规则，然后通过SQL语义来确定是否能将过滤条件前推到尽量早的时候执行。如果有子查询，过滤条件可以递归前推入最低层的子查询中，从而保证所有的冗余数据被删除。

超宽表的读取过滤

对一些列超多的表进行处理的时候，TranswarpInceptor首先会根据SQL语义来确定要读取的列，并在读取表的时候进行跨列读取减少IO和内存消耗。而如果表有过滤条件，Inceptor会做进一步优化，首先只读取过滤条件相关的列来确定该行记录是否需要被选择，如果不是就跳过当前行的所有列，因此能够最大程度上的减少数据读取。在一些商业实施中，这些优化规则能够带来5x-10x的性能提升。

Shuffle Stage的优化与消除

Spark的shuffle实现的效率非常低，需要把结果写磁盘，然后通过HTTP传输。TranswarpInceptor添加了一些shuffle消除的优化规则，对SQL的DAG中不必要或者是可以合并的shufflestage进行消除或者合并。对于必须要做Shuffle的计算任务，Inceptor通过DAGScheduler来提高shuffle的效率：MapTask会直接将结果返回给DAGScheduler，然后DAGScheduler将结果直接交给Reduce Task而不是等待所有Map Task结束，这样能够非常明显的提升shuffle阶段的性能。

Partition消除

TranswarpInceptor提供单一值Partition和RangePartition，并且支持对Partition建Bucket来做多次分区。当Partition过多的时候，系统的性能会因为内存消耗和调度开销而损失。因此，Inceptor提供了多个规则用于消除不必要的Partition，如果上下文中有隐式的对Partition的过滤条件，Inceptor也会生成对partition的过滤规则。

3.2.2基于成本的优化器(CostBasedOptimizer)

基于规则的优化器都是根据一些静态的信息来产生的，因此很多和动态数据相关的特性是不能通过基于规则的优化来解决，因此TranswarpInceptor提供了基于成本的优化器来做二次优化。相关的原始数据主要来自Meta-store中的表统计信息、RDD的信息、SQL上下文中的统计信息等。依赖于这些动态的数据，CBO会计算执行计划的物理成本并选择最有效的执行计划。一些非常有效的优化规则包括如下几点：

JOIN顺序调优

在实际的案例中，join是消耗计算量最多的业务，因此对join的优化至关重要。在多表JOIN模型中，TranswarpInceptor会根据统计信息来预估join的中间结果大小，并选择产生中间数据量最小的join顺序作为执行计划。

JOIN类型的选择

TranswarpInceptor支持Left-mostJoinTree 和 Bush Join Tree，并且会根据统计信息来选择生成哪种Join模型有最佳性能。此外，Transwarp Inceptor会根据原始表或者中间数据的大小来选择是否开启针对数据倾斜模型下的特殊优化等。此外，针对HBase表是否有索引的情况，Transwarp Inceptor会在普通Join和Look-up Join间做个均衡的选择。

并发度的控制

Spark通过线程级并发来提高性能，但是大量的并发可能会带来不必要的调度开销，因此不同的案例在不同并发度下会有最佳性能。TranswarpInceptor通过对RDD的一些属性进行推算来选择最佳并发控制，对很多的案例有着2x-3x的性能提升。

4.TranswarpHolodesk内存计算引擎

为了有效的降低SQL分析的延时，减少磁盘IO对系统性能的影响，星环科技研发了基于内存或者SSD的存储计算引擎TranswarpHolodesk，通过将表数据直接建在内存或者SSD上以实现SQL查询全内存计算。另外TranswarpHolodesk增加了数据索引功能，支持对多个数据列建索引，从而更大程度的降低了SQL查询延时。

4.1存储格式

TranswarpHolodesk基于列式存储做了大量的原创性改进带来更高的性能和更低的数据膨胀率。首先数据被序列化后存储到内存或SSD上以节省者资源占用。如图3所示，每个表的数据被存储成若干个Segment，每个Segment被划分成若干个Block，每个Block按照列方式存储于SSD或内存中。另外每个Block的头部都加上Min-MaxFilter和BloomFilter用于过滤无用的数据块，减少不必要的数据进入计算阶段。

TranswarpHolodesk根据查询条件的谓词属性对每个数据块的对应列构建数据索引，索引列采用自己研发的Trie结构进行组织存储，非索引列采用字典编码的方式进行组织存储。Trie不仅能对具有公共前缀的字符串进行压缩，而且可以对输入的字符串排序，从而可以利用二分查找快速查询所需数据的位置，从而快速响应查询需求。

HDFS2.6支持StorageTier让应用程序可以选择存储层为磁盘或者SSD，但是没有专用的存储格式设计是无法有效利用SSD的读写吞吐量和低延，因此现有的Text以及行列混合(ORC/Parquet)都不能有效的利用SSD的高性能。为此验证存储结构对性能的影响，我们将HDFS构建在SSD上并选用某基准测试来做了进一步的性能对比，结果如图4所示：采用文本格式，PCI-ESSD带来的性能提升仅1.5倍;采用专为内存和SSD设计的Holodesk列式存储，其性能相比较SSD上的HDFS提升高达6倍。

4.2性能优势

某运营商客户在12台x86服务器上搭建了TranswarpInceptor，将TranswarpHolodesk配置在PCIE-SSD上，并与普通磁盘表以及DB2来做性能对比测试。最终测试数据如图5所示：

在纯粹的count测试一项，Holodesk性能相对于磁盘表最高领先32倍;对于join测试一项，TranswarpHolodesk最高领先磁盘表多达12倍;在单表聚合测试中，Holodesk提升倍数达10~30倍。另外TranswarpHolodesk在和DB2的对比中也表现优秀，两个复杂SQL查询在DB2数据库中需要运行1小时以上，但是在使用TranswarpHolodesk均是分钟级和秒级就返回结果。

内存的价格大约是同样容量SSD的十倍左右，为了给企业提供更高性价比的计算方案，TranswarpHolodesk针对SSD进行了大量的优化，使得应用在SSD上运行具有与在内存上比较接近的性能，从而为客户提供了性价比更高的计算平台。

在对TPC-DS的IO密集型查询的测试中，无论上构建在PCI-ESSD还是内存上，Holodesk对比磁盘表有一个数量级上的性能提升;而SSD上的Holodesk性能只比内存差10%左右。

5.稳定的Spark执行引擎

企业目前应用开源Spark的主要困难在稳定性、可管理性和功能不够丰富上。开源Spark在稳定性上还有比较多的问题，在处理大数据量时可能无法运行结束或出现Outofmemory，性能时快时慢，有时比Map/Reduce更慢，无法应用到复杂数据分析业务中。

TranswarpInceptor针对各种出错场景设计了多种解决方法，如通过基于成本的优化器选择最合适的执行计划、加强对数据结构内存使用效率的有效管理、对常见的内存出错问题通过磁盘进行数据备份等方式，极大提高了Spark功能和性能的稳定性，上述问题都已经解决并经过商业案例的考验。TranswarpInceptor能稳定的运行7*24小时，并能在TB级规模数据上高效进行各种稳定的统计分析。

6.SQL引擎效能验证

TPC-DS是TPC组织为DecisionSupportSystem设计的一个测试集，包含对大数据集的统计/报表生成/联机查询/数据挖掘等复杂应用，测试用的数据有各种不同的分布与倾斜，与真实场景非常接近。随着国内外各代表性的Hadoop发行版厂商以TPC-DS为标准测评产品，TPC-DS也就逐渐成为了业界公认的Hadoop系统测试准则。

6.1验证对比的平台和配置

我们搭建了两个集群分别用于TranswarpInceptor与ClouderaDataHub/Impala的测试。

6.2TranswarpInceptorVS Cloudera Impala

TranswarpInceptor由于有完善的SQL支持，能够运行全部所有的99个SQL查询。而由于Cloudera官方发布的TPC-DS测试集只包含19个SQL案例，因此我们只能运行这19个SQL，实验证明这部分查询在Impala上全部正常运行完成。

6.3TranswarpInceptorVS Map Reduce

我们使用了同样的硬件和软件配置完成和开源的Hive执行效率相比，TranswarpInceptor能够带来10x-100x的性能提升。图8是TPC-DS的部分SQL查询在Inceptor和CDH5.1Hive的性能提升倍数，其中最大的提升倍数竟可达到123倍。

7.结语

随着在大数据领域国内外开始处于同一起跑线，我们相信像星环科技这样国内具有代表性的Hadoop发行版厂商将在中国的广阔市场空间中获得长足发展，并且由于中国市场激烈的竞争与磨练，逐步打磨出超越国外先进厂商的技术与实力。

刘汪根。2013年加入星环，作为早期员工参与了星环大数据平台的构建，现担任数据平台部研发经理，主要负责与管理星环大数据平台数据平台的研发工作，如SQL编译器，Spark执行引擎等工作，产品涵括TranswarpInceptor/TranswarpStream等软件。

【编者按】星环科技从2013年6月开始研发基于Spark的SQL执行引擎，在2013年底推出TranswarpInceptor1.0，并落地了国内首个7x24小时的商用项目。经过1年多的持续创新与改进，星环已经在国内落地了数十个Inceptor的商用项目。这是一篇星环Spark解决方案的技术解析，也是Spark用户可以效仿的优化之道。

大数据技术中，关于用户行为分析方面的有哪些技术

做用户行为分析的基础是获得用户行为数据，例如用户页面停留时间、跳转来源等等。这些信息有些能直接拿到，有些是需要做一些计算才能拿到的。一般来说用户访问时的一些信息都是以日志的形式打到web容器的日志空间中去，这其中包含基于spark服务行业的用户行为分析了最通用的一些访问信息以及一些自定义的日志打点。

题主提到了大数据技术中对用户行为进行分析，那么可以假定网站或者App的访问量是比较傲多的。由于系统流量比较大，计算维度又比较多，后续数据消费者的需求增长比较快，所以对计算分析平台有了一定的要求。具体表现为基于spark服务行业的用户行为分析：

1.负载能力。流量增大以后带来的压力是多方面的，比如网络带宽的压力、计算复杂度带来的压力、存储上的压力等等。一般来说这些都是比较显而易见的，会对产生比较直接的影响，比如计算实时性下降、消息出现了堆积、OOM等等。为了解决这一现象，一般来说会选择一些分布式的框架来解决这个问题，比如引入分布式计算框架storm、spark，分布式文件系统hdfs等。

2.实时性。在系统资源捉襟见肘时消息的实时性会立即受到严重影响，这使得部分算法失效（例如对计算和收集上来的数据进行行为分析后，反馈到推荐系统上，当整体响应时间过场时会严重影响推荐效果和准确度）。对于这个情况来说可能会选择storm这种具有高实时性的分布式流式计算框架来完成任务。

3.系统管理和平台化相关技术手段。在大数据情景下，企业内数据环境和应用环境都是比较复杂的，用户行为分析应用不是一成不变的，那么就要求用户行为分析这种多变的应用在复杂环境中能有效生存，这包括算法数据材料的获得、系统运维、系统任务调度、系统资源调度等等，相关的技术很多时候要求团队自研，但也有ganglia、yarn、mesos这类开源系统可以参考或者直接使用。

4.数据链路。企业技术环境一般来说是非常复杂的，一层一层交错在一起，远不是一句MVC三层架构能够概括得了的，为了避免消息流通呈复杂的网状结构，一般会考虑应用服务化、企业服务总线（ESB）及消息总线来做传输，有兴趣的话题主可以百度一下这几个方向的技术和开源工具。

5.应用快速生成工具。基于spark服务行业的用户行为分析我个人认为在大数据环境下应用都摆脱不了一个快速开发的要求，用户行为分析也是如此，这时候要考虑对接一些开源的分布式数据分析算法库而不是通过自己去实现，比如像spark ml，mahout这类的库用得好能减少很多工作量。