【发布时间】:2015-08-19 12:23:09
【问题描述】:
我的环境使用 Spark、Pig 和 Hive。
我在用 Scala(或与我的环境兼容的任何其他语言)编写可以将文件从本地文件系统复制到 HDFS 的代码时遇到了一些麻烦。
有人对我应该如何进行有任何建议吗?
【问题讨论】:
标签: scala hadoop apache-spark hive apache-pig
我的环境使用 Spark、Pig 和 Hive。
我在用 Scala(或与我的环境兼容的任何其他语言)编写可以将文件从本地文件系统复制到 HDFS 的代码时遇到了一些麻烦。
有人对我应该如何进行有任何建议吗?
【问题讨论】:
标签: scala hadoop apache-spark hive apache-pig
其他答案对我不起作用,所以我在这里再写一个。
试试下面的 Scala 代码:
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.fs.Path
val hadoopConf = new Configuration()
val hdfs = FileSystem.get(hadoopConf)
val srcPath = new Path(srcFilePath)
val destPath = new Path(destFilePath)
hdfs.copyFromLocalFile(srcPath, destPath)
您还应该检查 Spark 是否在 conf/spark-env.sh 文件中设置了 HADOOP_CONF_DIR 变量。这将确保 Spark 能够找到 Hadoop 配置设置。
build.sbt 文件的依赖关系:
libraryDependencies += "org.apache.hadoop" % "hadoop-common" % "2.6.0"
libraryDependencies += "org.apache.commons" % "commons-io" % "1.3.2"
libraryDependencies += "org.apache.hadoop" % "hadoop-hdfs" % "2.6.0"
【讨论】:
您可以使用 Hadoop FileSystem API 编写 Scala 作业。
并使用 apache commons 中的IOUtils 将数据从 InputStream 复制到 OutputStream
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.commons.io.IOUtils;
val hadoopconf = new Configuration();
val fs = FileSystem.get(hadoopconf);
//Create output stream to HDFS file
val outFileStream = fs.create(new Path("hedf://<namenode>:<port>/<filename>))
//Create input stream from local file
val inStream = fs.open(new Path("file://<input_file>"))
IOUtils.copy(inStream, outFileStream)
//Close both files
inStream.close()
outFileStream.close()
【讨论】:
libraryDependencies += "org.apache.hadoop" % "hadoop-common" % "2.6.0"libraryDependencies += "org.apache.commons" % "commons-io" % "1.3.2"libraryDependencies += "org.apache.hadoop" % "hadoop-hdfs" % "2.6.0"
这是适用于 S3 的东西(从上面修改)
def cpToS3(localPath: String, s3Path: String) = {
val hdfs = FileSystem.get(
new URI(s3Path),
spark.sparkContext.hadoopConfiguration)
val srcPath = new Path(localPath)
val destPath = new Path(s3Path)
hdfs.copyFromLocalFile(srcPath, destPath)
}
【讨论】: