不论是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以及以后才会支持的。
spark进入txt文件的命令1、首先启动spark-shell进入Spark-shell模式：（进入spark目录下后 输入命令 bin/spark-shell启动spark-shell模式）
2、加载text文件（spark创建sc，可以加载本地文件和HDFS文件戚庆州创建RDD）
val textFile = sc.textFile("file:///home/hadoop/test1.txt") #注意file:后是三个“/”
注意：加载HDFS文件和本地文件都是使用textFile，区别是添加前缀(hdfs://和file://)进行标识。
3、获取RDD文件textFile所有项（文本文件即总共行数）的计数（还有很多其他的RDD *** 作，自行百度）
textFile.count() #统计结果显示 1 行
二、在 spark-shell 中读取 HDFS 系统文件“/home/hadoop/test.csv（也可以是txt文件）”（如果该差告文件不存在， 请先创建），然后，统计出文件的行数：
方法一：
1、加载text文件（spark创建sc，可以加载本地文件和HDFS文件创建RDD）
val textFile = sc.textFile("hdfs:///home/hadoop/test.csv") #注意hdfs:后是三个“/”
注意：加载HDFS文件和本地文件都是使用textFile，区别是添加前缀(hdfs://和file://)进行标识。
2、获取RDD文件textFile所有项的计数
textFile.count() #统计结果显示 1 行
方法二：（Spark shell 默认是读取 HDFS 中的文件，需要先上传文件到 HDFS 中，否则会有“org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: hdfs://localhost:9000/user/hadoop/README.md”的错误。）
1、省去方法一中第一步的命令（1）中的“hdfs：//”，其他部分相同，命令如下：
三、编写独立应用程序，读取 HDFS 系统文件“/user/hadoop/test.txt”（如果该文件不存在， 请先创建），然后，统计出文件的行数；通过 sbt 工具将整个应用程序编译打包成 JAR 包， 并将生成的 JAR 包通过 spark-submit 提交到 Spark 中运行命令：
1、首先输入:quit 命令退出spark-shell模式：
2、在终端中执行如下命令创建一个文件夹 sparkapp3 作为应用程序根目录：
cd ~ # 进入用户主文件夹
mkdir ./sparkapp3 # 创建应用程序根目录
mkdir -p ./sparkapp3/src/高蔽main/scala # 创建所需的文件夹结构
3、在 ./sparkapp3/src/main/scala 下建立一个名为 SimpleApp.scala 的文件（vim ./sparkapp3/src/main/scala/SimpleApp.scala），添加代码如下：
/* SimpleApp.scala */
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
object SimpleApp {
def main(args: Array[String]) {
val logFile = "hdfs://localhost:9000/home/hadoop/test.csv"
val conf = new SparkConf().setAppName("Simple Application")
val sc = new SparkContext(conf)
val logData = sc.textFile(logFile, 2)
val num = logData.count()
println("这个文件有 %d 行！".format(num))
}
}
4、该程序依赖 Spark API，因此我们需要通过 sbt 进行编译打包。 ./sparkapp3 中新建文件 simple.sbt（vim ./sparkapp3/simple.sbt），添加内容如下，声明该独立应用程序的信息以及与 Spark 的依赖关系：
name := "Simple Project"
version := "1.0"
scalaVersion := "2.12.10"
libraryDependencies += "org.apache.spark" %% "spark-core" % "3.0.0-preview2"
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）