使用 Scala 将文件写入 HDFS

Question

我正在尝试使用 scala File System Api 将文件写入 HDFS，在客户端上出现以下错误，在 hadoop 日志上也出现同样的错误：

文件 /user/testuser/test.txt 只能复制到 0 个节点 而不是 minReplication (=1)。有 1 个数据节点正在运行，1 个 此操作中排除节点。

testuser 有读、写和执行的权限。我检查了 ambari 上的 hdfs 是否已启动并正在运行，不知道为什么会出现此错误

在为错误做谷歌之后，我已经尝试停止所有服务，格式化名称节点并启动所有服务等，就像它在下面的链接中所说的那样

Writing to HDFS could only be replicated to 0 nodes instead of minReplication (=1)

我仍然有同样的错误。任何建议我做错了什么，我是 hadoop 新手，所以任何建议都将不胜感激。

以下是我正在使用的 scala 代码

def write(uri: String, filePath: String, data: Array[Byte]) = {
        System.setProperty("HADOOP_USER_NAME", "usernamehere")
        val path = new Path(filePath)
        val conf = new Configuration()
        conf.set("fs.defaultFS", uri)
        conf.set("dfs.client.use.datanode.hostname", "true");
        conf.addResource(new Path("/path/core-site.xml"));
        conf.addResource(new Path("/path/hdfs-site.xml"));
        val fs = FileSystem.get(conf)
         val os = fs.create(path)
        fs.setPermission(path,FsPermission.getDefault)
        val out = new BufferedOutputStream(os)
        println(data.length)
        out.write(data)
        out.flush()
        out.close()
        fs.close()

      }

谢谢

Answer 1

要将任何文件写入 HDFS，您只需要使用 hdfs 命令，如 copyFromLocal。

假设您使用的是带有 scala 的 Spark，那么您需要使用 Spark 文件写入命令，例如。

some_dataframe.write.mode(SaveMode.Overwrite).parquet("c:\\MyWorkSpace\\Spark\\")

HDFS 理解上述命令以根据复制因子复制数据。但是，如果您在 HDFS 中使用 scala 文件系统 api，则会导致问题，因为 scala 无法理解 hdfs 的功能，如复制、数据块、分区。