Выполнение команд hdfs из скрипта scala

Question

Я пытаюсь выполнить команду HDFS изнутри скрипта scala, выполняемого Spark в режиме кластера.Ниже команды:

val cmd = Seq("hdfs","dfs","-copyToLocal","/tmp/file.dat","/path/to/local")
val result = cmd.!!

На этом этапе задание не выполняется с ошибкой, как показано ниже:

java.io.FileNotFoundException: /var/run/cloudera-scm-agent/process/2087791-yarn-NODEMANAGER/log4j.properties (Permission denied)
        at java.io.FileInputStream.open0(Native Method)
        at java.io.FileInputStream.open(FileInputStream.java:195)
        at java.io.FileInputStream.<init>(FileInputStream.java:138)
        at java.io.FileInputStream.<init>(FileInputStream.java:93)
        at sun.net.www.protocol.file.FileURLConnection.connect(FileURLConnection.java:90)
        at sun.net.www.protocol.file.FileURLConnection.getInputStream(FileURLConnection.java:188)
        at org.apache.log4j.PropertyConfigurator.doConfigure(PropertyConfigurator.java:557)
        at org.apache.log4j.helpers.OptionConverter.selectAndConfigure(OptionConverter.java:526)
        at org.apache.log4j.LogManager.<clinit>(LogManager.java:127)
        at org.apache.log4j.Logger.getLogger(Logger.java:104)
        at org.apache.commons.logging.impl.Log4JLogger.getLogger(Log4JLogger.java:262)
        at org.apache.commons.logging.impl.Log4JLogger.<init>(Log4JLogger.java:108)
        at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
        at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
        at java.lang.reflect.Constructor.newInstance(Constructor.java:423)

Однако, когда я запускаю ту же команду отдельно в оболочке Spark, она выполняется простохорошо, и файл также копируется.

scala> val cmd = Seq("hdfs","dfs","-copyToLocal","/tmp/file_landing_area/file.dat","/tmp/local_file_area")
cmd: Seq[String] = List(hdfs, dfs, -copyToLocal, /tmp/file_landing_area/file.dat, /tmp/local_file_area)

scala> val result = cmd.!!
result: String = ""

Я не понимаю, ошибка отказа в разрешении.Хотя отображается как FileNotFoundException.Совершенно запутанно.

Есть идеи?

Answer 1

В соответствии с ошибкой, он проверяет данные hdfs в папке var, что, как я подозреваю, является проблемой конфигурации или не указывает на правильную.Использование seq и выполнение команды HDFS не является хорошей практикой.Это полезно только для искровой оболочки.Использование того же подхода в коде не рекомендуется.Вместо этого попробуйте использовать приведенный ниже API файловой системы Scala для перемещения данных из или в HDFS.Пожалуйста, проверьте приведенный ниже пример кода только для справки, которая может вам помочь.

import org.apache.hadoop.fs
import org.apache.hadoop.fs._
val conf = new Configuration()

val fs = path.getFileSystem(conf)

val hdfspath = new Path("hdfs:///user/nikhil/test.csv")
val localpath = new Path("file:///home/cloudera/test/")

fs.copyToLocalFile(hdfspath,localpath)

Пожалуйста, используйте ссылку ниже для получения дополнительной ссылки относительно API файловой системы Scala.

https://hadoop.apache.org/docs/r2.9.0/api/org/apache/hadoop/fs/FileSystem.html#copyFromLocalFile(boolean,%20boolean,%20org.apache.hadoop.fs.Path,%20org.apache.hadoop.fs.Path)