不论是Hive还是Spark SQL在使用过程中都可能会遇到小文件过多的问题。小文件过多最直接的表现是任务执行时间长,查看Spark log会发现大量的数据移动的日志。我们可以查看log中展现的日志信息,去对应的路径下查看文件的大小和个数。租敬稿
通过上述命令可以查看文件的个数以及大小。count查看出的文件大小单位是B,需要转换为MB。
在spark官方的推荐文档中,parquet格式的文件推荐大小是128MB,小于该大小的均可以称之为小文件,在实际的工作,往往小文件的大小仅仅为几KB,表现为,可能文件大小为几百MB,但是文件个数可能到达了几十万个。一般来说,我们可以通过简单相除获得文件的平均大小,如果文件数目不多,我们也可以通过下述命令获得每个文件的大小。
1.任务执行时间长
2.真实的文件大小独占一个数据存储块,存放到DataNode节点中。同时 DataNode一般默认存三份副本,以保障数据安全。同时该文件所存放的位置也写入到NameNode的内存中,如果有Secondary NameNode高可用节点,也可同时复制一份过去。NameNode的内存数据将会存放到硬盘中,如果HDFS发生重弊孝启,将产生较长时间的元数据从硬盘读到内存的过程。
3.不论在Hive还是在Spark中,每一个存储块都对应一个Map程序,一个Map呈现就需要一个JVM,启动一个JVM去读取或者写小文件是吃力不讨好的行为。在实际的生产中,为了更好的管理集群资源,一般会要求程序执行时限制Executor数量和每个Executor的核心数量,需要频繁创建Executor来读取写入。
5.影响磁盘寻址时间
小文件合并,本质上就是通过某种 *** 作,将一系列小文件合并成大文件。我们知道,以MapReduce为代表的大数据系统,都习惯用K-V键值对的形式来处理文件,最后文件落盘,也是一个reduce对应一个输出文件。所以直观上,我们可以减少reduce数量,达到减少文件数量的目的。
从Map到Reduce需要一个Shuffle过程,所以我们将小文件合并理解为通过一个Shuffle,合并小文件成一个大文件。基于这样的思想,我们的策略可以稿耐分为两类:一类是原来的计算已经有Shuffle了,那么我们可以认为控制输出文件的数量;二类是强制触发Shuffle,进行小文件合并。
1-设置参数 (一般用于Hive)
2-distribute by rand()
往动态分区插入数据时,在已经写好的SQL末尾加上distribute by rand()
该算子只是起到打散的效果,但是我们还要设置文件的大小,以免打散后仍然有小文件。
表示每个reduce的大小,Hive可以数据总量,得到reduce个数,假设hive认为会有10个reduce,那么,这里rand()则会为 x % 10
3-group by
我们知道,group by算子会触发Shuffle,因此只要我们设置好Shuffle时的文件个数就好,在Spark SQL中,我们可以设置partition个数,因为一个partition会对应一个文件。
上述的 *** 作,会触发shuffle,因此我们再设置partition个数。
则表示,shuffle后,只会产生10个partition.
4-repartition()
5-coalesce()
需要注意的是,4和5都是spark 2.4以及以后才会支持的。
一、hadoop、Storm该选哪一个?
为了区别hadoop和Storm,该部分将回答如下问题:
1.hadoop、Storm各是什么运算
2.Storm为什么被称之为流式计算系统
3.hadoop适合什么场景,什么情况下使用hadoop
4.什么是吞吐量
首先整体认识:Hadoop是级计算,进行计算时,数据在磁盘上,需要读写磁盘;Storm是内存级计算,数据直接通过网络导入内存。读写内存比读写磁盘速度快n个数量级。根据Harvard CS61课件,磁盘访问延迟约为内存访问延迟的75000倍。所以Storm更快。
注释:
1. 延时 , 指数据从产生到运算产生结果的时间,“快”应该主要指这个。
2. 吞吐, 指系统单位时间处理的数据量。
storm的网络直传、内存计算,其时延必然比hadoop的通过hdfs传输低得多;当计算模型比较适合流式时,storm的流式处理,省去了批处理的收集数据的时间;因为storm是服务型的作业,也省去了作业调度的时延。所以从时延上来看,storm要快于hadoop。
从原理角度来讲:
Hadoop M/R基于HDFS,需要切分输入数据、产生中间数据文件、排序、数据压缩、多份复制等,效率较低。
Storm 基于ZeroMQ这个高性能的消息通讯库,不持久化数据。
为什么storm比hadoop快,下面举一个应用场景
说一个典型的场景,几千个日志生产方产生日志文件,需要进行一些ETL *** 作存入一个数据库。
假设利用hadoop,则需要先存入hdfs,按每一分钟切一个文件的粒度来算(这个粒度已经极端的细了,再小的拍颂话hdfs上会一堆小文件),hadoop开始计算时,1分钟已经过去了,然后再开始调度任务又花了一分钟,然后作业运行起来,假设机器特别多,几钞钟就算完了,然后写数据库假设也花了很少的时间,这样,从数据产生到最后可以使用已经过去了至少两分多钟。
而流式计算则是数据产生时,则有一个程序去一直监控日志的产生,产生一行就通过一个传输系统发给流式计算系统,然后流式计算系统直接处理,处理完之后直接写入,每条数据从产生到写入数据库,在资源充足时可以在毫秒败升级别完成。
同时说一下另外一个场景:
如果一个大文件的wordcount,把它放到storm上进行流式的处理,等所有已有数据处理完才让storm输出结果,这时候,你再把它和hadoop比较快慢,这时,其实比较的不是时延,而是比较的吞吐了。
--------------------------------------------------------------------------------------------------------------------------------
最主要的方面:Hadoop使用作为中间交换的介质,而storm的数据是一直在内存中流转的。
两者面向的领域也不完全相同,一个是批量处理,基于任务调度的;另外一个是实时处理,基于流。
以水为例,Hadoop可以看作是纯净水,一桶桶地搬;而Storm是用水管,预先接好(Topology),然后打开水龙头,水就源源不断地流出来了。
--------------------------------------------------------------------------------------------------------------------------------
Storm的主工程师Nathan Marz表示: Storm可袭枯郑以方便地在一个计算机集群中编写与扩展复杂的实时计算,Storm之于实时处理,就好比Hadoop之于批处理。Storm保证每个消息都会得到处理,而且它很快——在一个小集群中,每秒可以处理数以百万计的消息。更棒的是你可以使用任意编程语言来做开发。
Storm的主要特点如下:
1.简单的模型。类似于MapReduce降低了并行批处理复杂性,Storm降低了进行实时处理的复杂性。
2.可以使用各种。你可以在Storm之上使用各种编程语言。默认支持Clojure、Java、Ruby和Python。要增加对其他语言的支持,只需实现一个简单的Storm通信协议即可。
3.容错性。Storm会管理工作进程和节点的故障。
4.水平扩展。计算是在多个线程、进程和服务器之间并行进行的。
5.可靠的消息处理。Storm保证每个消息至少能得到一次完整处理。任务失败时,它会负责从消息源重试消息。
6.快速。系统的设计保证了消息能得到快速的处理,使用MQ作为其底层消息队列。
7.本地模式。Storm有一个“本地模式”,可以在处理过程中完全模拟Storm集群。这让你可以快速进行开发和单元测试。
--------------------------------------------------------------------------------------------------------------------------------
在消耗资源相同的情况下,一般来说storm的延时低于mapreduce。但是吞吐也低于mapreduce。storm是典型的流计算系统,mapreduce是典型的批处理系统。下面对流计算和批处理系统流程
这个个数据处理流程来说大致可以分三个阶段:
1. 数据采集与准备
2. 数据计算(涉及计算中的中间存储), 题主中的“那些方面决定”应该主要是指这个阶段处理方式。
3. 数据结果展现(反馈)
1)阶段,目前典型的处理处理策略:数据的产生系统一般出自页面打点和解析DB的log,流计算将数据采集中消息队列(比如kafaka,metaQ,timetunle)等。批处理系统一般将数据采集进分布式文件系统(比如HDFS),当然也有使用消息队列的。我们暂且把消息队列和文件系统称为预处理存储。二者在延时和吞吐上没太大区别,接下来从这个预处理存储进入到数据计算阶段有很大的区别,流计算一般在实时的读取消息队列进入流计算系统(storm)的数据进行运算,批处理一系统一般会攒一大批后批量导入到计算系统(hadoop),这里就有了延时的区别。
2)数据计算阶段,流计算系统(storm)的延时低主要有一下几个方面(针对题主的问题)
A: storm 进程是常驻的,有数据就可以进行实时的处理
mapreduce 数据攒一批后由作业管理系统启动任务,Jobtracker计算任务分配,tasktacker启动相关的运算进程
B: stom每个计算单元之间数据之间通过网络(zeromq)直接传输。
mapreduce map任务运算的结果要写入到HDFS,在于reduce任务通过网络拖过去运算。相对来说多了读写,比较慢
C: 对于复杂运算
storm的运算模型直接支持DAG(有向无环图)
mapreduce 需要肯多个MR过程组成,有些map *** 作没有意义的
3)数据结果展现
流计算一般运算结果直接反馈到最终结果集中(展示页面,,搜索引擎的索引)。而mapreduce一般需要整个运算结束后将结果批量导入到结果集中。
实际流计算和批处理系统没有本质的区别,像storm的trident也有批概念,而mapreduce可以将每次运算的数据集缩小(比如几分钟启动一次),facebook的puma就是基于hadoop做的流计算系统。
二、高性能并行计算引擎Storm和Spark比较
Spark基于这样的理念,当数据庞大时,把计算过程传递给数据要比把数据传递给计算过程要更富效率。每个节点存储(或缓存)它的数据集,然后任务被提交给节点。
所以这是把过程传递给数据。这和Hadoop map/reduce非常相似,除了积极使用内存来避免I/O *** 作,以使得迭代算法(前一步计算输出是下一步计算的输入)性能更高。
Shark只是一个基于Spark的查询引擎(支持ad-hoc临时性的分析查询)
而Storm的架构和Spark截然相反。Storm是一个分布式流计算引擎。每个节点实现一个基本的计算过程,而数据项在互相连接的网络节点中流进流出。和Spark相反,这个是把数据传递给过程。
两个框架都用于处理大量数据的并行计算。
Storm在动态处理大量生成的“小数据块”上要更好(比如在Twitter数据流上实时计算一些汇聚功能或分析)。
Spark工作于现有的数据全集(如Hadoop数据)已经被导入Spark集群,Spark基于in-memory管理可以进行快讯扫描,并最小化迭代算法的全局I/O *** 作。
不过Spark流模块(Streaming Module)倒是和Storm相类似(都是流计算引擎),尽管并非完全一样。
Spark流模块先汇聚批量数据然后进行数据块分发(视作不可变数据进行处理),而Storm是只要接收到数据就实时处理并分发。
不确定哪种方式在数据吞吐量上要具优势,不过Storm计算时间延迟要小。
总结下,Spark和Storm设计相反,而Spark Steaming才和Storm类似,前者有数据平滑窗口(sliding window),而后者需要自己去维护这个窗口。
spark和hadoop的区别:诞生的先后顺序、计算不同、平台不同。诞生的先后顺序,hadoop属于第一代开源大数据处理平台,而spark属于第二森盯代。属于下一代的spark肯定在综合评价上要优于第一代的hadoop。
计算不同spark和hadoop在分布式计算的底层思路上,其实宏昌是极为相似的,即mapreduce分布式运算模此绝和型:将运算分成两个阶段,阶段1-map,负责从上游拉取数据后各自运算,然后将运算结果shuffle给下游的reduce,reduce再各自对通过shuffle读取来的数据进行聚合运算spark和hadoop在分布式计算的具体实现上,又有区别;hadoop中的mapreduce运算框架,一个运算job,进行一次map-reduce的过程;而spark的一个job中,可以将多个map-reduce过程级联进行。
平台不同spark和hadoop区别是,spark是一个运算平台,而hadoop是一个复合平台(包含运算引擎,还包含分布式文件存储系统,还包含分布式运算的资源调度系统),所以,spark跟hadoop来比较的话,主要是比运算这一块大数据技术发展到目前这个阶段,hadoop主要是它的运算部分日渐式微,而spark目前如日中天,相关技术需求量大,offer好拿。
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)