Spark读取和存储HDFS上的数据
本篇来介绍一下通过Spark来读取和HDFS上的数据，主要包含四方面的内容：将RDD写入HDFS、读取HDFS上的文件、将HDFS上的文件添加到Driver、判断HDFS上文件路径是否存在。
本文的代码均在本地测试通过，实用的环境时MAC上安装的Spark本地环境。
1、启动Hadoop
首先启动咱们的Hadoop，在hadoop的目录下执行下面的命令：
rm -rf tmp
mkdir tmp
cd sbin
hadoop namenode -format
start-dfs.sh
start-yarn.sh查看是否启动成功：
2、将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:
3、读取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，则必须加上。
4、将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,DIEN5、判断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而在公司中的大规模集群环境下，通常的代码如下：
val conf = spark.sparkContext.hadoopConfiguration
val fs = org.apache.hadoop.fs.FileSystem.get(conf)
val modelNamesExists = fs.exists(new org.apache.hadoop.fs.Path("hdfs://localhost:9000/user/root/modelNames/part-00000"))
val modelNames1Exists = fs.exists(new org.apache.hadoop.fs.Path("hdfs://localhost:9000/user/root/modelNames1/part-00000"))
println(modelNamesExists)
println(modelNames1Exists)如果在本地单机环境下仍然使用上面的代码，会报如下的错误：
Wrong FS: hdfs://localhost:9000/user/root/modelNames/part-00000, expected: file:///所以对比两份代码你可以发现，在本地环境中，我们首先使用getFileSystem获取了hdfs文件系统中的路径信息，从而避免了上面的错误。
好了，今天的知识就分享到这里，小伙伴们都掌握了么？
腾讯云开发者
