Spark写数据到Hadoop：HDFS与 Spark如何协同工作

Spark作为大数据计算框架，主要的优势在于数据计算，但是因为自身不具备分布式文件系统，所以往往需要结合其他的文件系统和数据源来完成工作，这其中就包括HDFS。HDFS是Hadoop的分布式文件系统，Spark写数据到Hadoop，其实就是HDFS与Spark协同工作，那么这一过程要如何实现呢？

Spark框架不管是写数据还是读取数据，基于Hadoop集群，主要靠的就是HDFS。我们可以把这个过程分为四个阶段：

将RDD写入HDFS→读取HDFS上的文件→将HDFS上的文件添加到Driver→判断HDFS上文件路径是否存在。


Spark写数据到Hadoop，其实就是将RDD写入HDFS，当面对数据处理需求，向HDFS请求数据，读取RDD，然后才能进行后续的数据处理环节。

第一步：启动Hadoop

首先启动Hadoop，在hadoop的目录下执行下面的命令：

rm -rf tmp

mkdir tmp

cd sbin

hadoop namenode -format

start-dfs.sh

start-yarn.sh

查看是否启动成功：

第二步：将RDD写入HDFS

先创建一个SparkSession：

val spark = SparkSession

      .builder()

      .appName("Spark SQL basic example")

      .enableHiveSupport()

      .getOrCreate()

    spark.sparkContext.setLogLevel("WARN")

将RDD写入HDFS使用的函数是saveAsTextFile：

val modelNames = Array("FM","FFM","DEEPFM","NFM","DIN","DIEN")

val modelNamesRdd = spark.sparkContext.parallelize(modelNames,1)

modelNamesRdd.saveAsTextFile("hdfs://localhost:9000/user/root/modelNames")

接下来，查看一下是否保存成功：

可以看到RDD在HDFS上是分块存储的，由于我们只有一个分区，所以只有part-0000。假设我们存储一个包含两个分区的RDD：

val modelNames3 = Array("FM","FFM","DEEPFM","NFM","DIN","DIEN")

val modelNames3Rdd = spark.sparkContext.parallelize(modelNames3,2)

modelNames3Rdd.saveAsTextFile("hdfs://localhost:9000/user/root/modelNames3")

再次查看，可以看到有part-00000和part-00001:

第三步：读取HDFS上的文件

读取HDFS上的文件，使用textFile方法：

val modelNames2 = spark.sparkContext.textFile("hdfs://localhost:9000/user/root/modelNames/part-00000")

val modelNames4 = spark.sparkContext.textFile("hdfs://localhost:9000/user/root/modelNames3/")

读取时是否加最后的part-00000都是可以的，当只想读取某个part，则必须加上。

第四步：将HDFS上的文件添加到Driver

有时候，我们并不想直接读取HDFS上的文件，而是想对应的文件添加到Driver上，然后使用java或者Scala的I／O方法进行读取，此时使用addFile和get方法来实现：

val files = "hdfs://localhost:9000/user/root/modelNames/part-00000"

spark.sparkContext.addFile(files)

val path = SparkFiles.get("part-00000")

println(path)

打印的路径十分奇怪（没有截取完全）：

然后有了path之后，就可以使用scala的I／O进行读取：

val source = Source.fromFile(path)

val lineIterator = source.getLines

val lines =lineIterator.toArray

println(lines.mkString(","))

输出为：

FM,FFM,DEEPFM,NFM,DIN,DIEN

第五步：判断HDFS上文件路径是否存在

在读取HDFS地址或者将文件传输到Driver上的时候，首先需要判断文件是否存在。单机环境下，代码如下：

val conf = spark.sparkContext.hadoopConfiguration

val path = new org.apache.hadoop.fs.Path("hdfs://localhost:9000/user/root/modelNames/part-00000")

val fs = path.getFileSystem(conf) //得hdfs文件系统中的路径信息

val modelNamesExists = fs.exists(path)

val modelNames1Exists = fs.exists(new org.apache.hadoop.fs.Path("hdfs://localhost:9000/user/root/modelNames1/part-00000"))

println(modelNamesExists)

println(modelNames1Exists)

输出结果为：

true

false

以上就是对Spark写数据到Hadoop这个过程的详细介绍了，当然这是基于单机环境来操作的，在真实的Hadoop集群环境当中，面临的数据以及细节可能会更复杂，但是基本流程就是这样完成的。成都加米谷大数据，大数据技术分享，线下大数据培训班，课程详情可联系客服了解！