如何使用 Spark 从 HDFS 读取文件？

Question

我已经使用 Apache Spark 构建了一个推荐系统，数据集本地存储在我的项目文件夹中，现在我需要从 HDFS 访问这些文件。

如何使用 Spark 从 HDFS 读取文件？

这就是我初始化 Spark 会话的方式：

SparkContext context = new SparkContext(new SparkConf().setAppName("spark-ml").setMaster("local")
                .set("fs.default.name", "hdfs://localhost:54310").set("fs.defaultFS", "hdfs://localhost:54310"));
        Configuration conf = context.hadoopConfiguration();
        conf.addResource(new Path("/usr/local/hadoop-3.1.2/etc/hadoop/core-site.xml"));
        conf.addResource(new Path("/usr/local/hadoop-3.1.2/etc/hadoop/hdfs-site.xml"));
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        conf.set("fs.hdfs.impl", "org.apache.hadoop.fs.LocalFileSystem");
        this.session = SparkSession.builder().sparkContext(context).getOrCreate();
        System.out.println(conf.getRaw("fs.default.name"));
        System.out.println(context.getConf().get("fs.defaultFS"));

所有输出都返回 hdfs://localhost:54310，这是我的 HDFS 的正确 uri。

尝试从 HDFS 读取文件时：

session.read().option("header", true).option("inferSchema", true).csv("hdfs://localhost:54310/recommendation_system/movies/ratings.csv").cache();

我得到这个错误：

Exception in thread "main" java.lang.IllegalArgumentException: Wrong FS: hdfs://localhost:54310/recommendation_system/movies/ratings.csv, expected: file:///
    at org.apache.hadoop.fs.FileSystem.checkPath(FileSystem.java:730)
    at org.apache.hadoop.fs.RawLocalFileSystem.pathToFile(RawLocalFileSystem.java:86)
    at org.apache.hadoop.fs.RawLocalFileSystem.deprecatedGetFileStatus(RawLocalFileSystem.java:636)
    at org.apache.hadoop.fs.RawLocalFileSystem.getFileLinkStatusInternal(RawLocalFileSystem.java:930)
    at org.apache.hadoop.fs.RawLocalFileSystem.getFileStatus(RawLocalFileSystem.java:631)
    at org.apache.hadoop.fs.FilterFileSystem.getFileStatus(FilterFileSystem.java:454)
    at org.apache.hadoop.fs.Globber.getFileStatus(Globber.java:65)
    at org.apache.hadoop.fs.Globber.doGlob(Globber.java:281)
    at org.apache.hadoop.fs.Globber.glob(Globber.java:149)
    at org.apache.hadoop.fs.FileSystem.globStatus(FileSystem.java:2034)
    at org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:257)
    at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:228)
    at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:313)
    at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:204)
    at org.apache.spark.rdd.RDD.$anonfun$partitions$2(RDD.scala:253)
    at scala.Option.getOrElse(Option.scala:138)
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:251)
    at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:49)
    at org.apache.spark.rdd.RDD.$anonfun$partitions$2(RDD.scala:253)
    at scala.Option.getOrElse(Option.scala:138)
    at org.apache.spark.rdd.RDD.partitions(RDD.scala:251)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2126)
    at org.apache.spark.rdd.RDD.$anonfun$collect$1(RDD.scala:945)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
    at org.apache.spark.rdd.RDD.collect(RDD.scala:944)
    at org.apache.spark.api.java.JavaRDDLike.collect(JavaRDDLike.scala:361)
    at org.apache.spark.api.java.JavaRDDLike.collect$(JavaRDDLike.scala:360)
    at org.apache.spark.api.java.AbstractJavaRDDLike.collect(JavaRDDLike.scala:45)
    at com.dastamn.sparkml.analytics.SparkManager.<init>(SparkManager.java:36)
    at com.dastamn.sparkml.Main.main(Main.java:22)

我能做些什么来解决这个问题？

Answer 1

sn-p 粘贴的代码中的一些内容：
1. 当一个hadoop属性必须设置为使用SparkConf的一部分时，它必须以spark.hadoop.为前缀，在这种情况下，fs.default.name需要设置为spark.hadoop.fs.default.name，其他属性也是如此。
2.csv函数的参数不必告诉HDFS端点，Spark会从默认属性中找出它，因为它已经设置好了。

session.read().option("header", true).option("inferSchema", true).csv("/recommendation_system/movies/ratings.csv").cache();

如果默认文件系统属性未在 HadoopConfiguration 中设置，则 Spark/Hadoop 需要完整的 URI 才能确定要使用的文件系统。
（也没有使用对象名称conf）
3. 在上述情况下，看起来 Hadoop 无法找到 hdfs:// URI 前缀的文件系统，并在这种情况下诉诸使用默认文件系统 local（因为它使用 RawLocalFileSystem 处理路径)。
确保hadoop-hdfs.jar 存在于具有DistributedFileSystem 的类路径中，以便为HDFS 设置FS 对象。

Answer 2

这是解决问题的配置：

SparkContext context = new SparkContext(new SparkConf().setAppName("spark-ml").setMaster("local[*]")
                .set("spark.hadoop.fs.default.name", "hdfs://localhost:54310").set("spark.hadoop.fs.defaultFS", "hdfs://localhost:54310")
                .set("spark.hadoop.fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName())
                .set("spark.hadoop.fs.hdfs.server", org.apache.hadoop.hdfs.server.namenode.NameNode.class.getName())
                .set("spark.hadoop.conf", org.apache.hadoop.hdfs.HdfsConfiguration.class.getName()));
        this.session = SparkSession.builder().sparkContext(context).getOrCreate();