yystopf
/
piflow
forked from opensci/piflow
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Optimization of HDFS file storage 

yang qidong
This commit is contained in:
yanfqidong0604 2018-12-13 22:48:15 +08:00
parent bcd658bb11
commit 0a4e2f24e7
2 changed files with 43 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
piflow-bundle/src/main/scala/cn/piflow/bundle/http/FileDownHDFS.scala
View File

@ -37,6 +37,31 @@ class FileDownHDFS extends ConfigurableStop{
val configuration: Configuration = new Configuration()
// "savePath":"hdfs://10.0.88.70:9000/yqd/hdfstest/dblp.xml.gz"
// val hdfsUrl = "hdfs://10.0.88.70:9000"
val pathARR: Array[String] = savePath.split("\\/")
var hdfsUrl:String=""
for (x <- (0 until 3)){
hdfsUrl+=(pathARR(x) +"/")
}
configuration.set("fs.defaultFS",hdfsUrl)
// configuration.set("dfs.nameservices", "nameservice1")
// configuration.set("dfs.ha.namenodes.nameservice1", "nn1,nn2");
// configuration.set("dfs.namenode.rpc-address.nameservice1.nn1", "xxx:8020");
// configuration.set("dfs.namenode.rpc-address.nameservice1.nn2", "xxx:8020");
// configuration.set("dfs.client.failover.proxy.provider.nameservice1"
// ,"org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
// configuration.addResource("classpath:/hadoop/core-site.xml");
// configuration.addResource("classpath:/hadoop/hdfs-site.xml");
// configuration.addResource("classpath:/hadoop/mapred-site.xml");
val fs = FileSystem.get(configuration)
val fdos: FSDataOutputStream = fs.create(new Path(savePath))

18
piflow-bundle/src/main/scala/cn/piflow/bundle/http/UnzipFilesOnHDFS.scala
View File

@ -27,6 +27,24 @@ class UnzipFilesOnHDFS extends ConfigurableStop {
val session: SparkSession = pec.get[SparkSession]()
val configuration: Configuration = new Configuration()
val pathARR: Array[String] = filePath.split("\\/")
var hdfsUrl:String=""
for (x <- (0 until 3)){
hdfsUrl+=(pathARR(x) +"/")
}
configuration.set("fs.defaultFS",hdfsUrl)
// configuration.set("dfs.nameservices", "nameservice1")
// configuration.set("dfs.ha.namenodes.nameservice1", "nn1,nn2");
// configuration.set("dfs.namenode.rpc-address.nameservice1.nn1", "xxx:8020");
// configuration.set("dfs.namenode.rpc-address.nameservice1.nn2", "xxx:8020");
// configuration.set("dfs.client.failover.proxy.provider.nameservice1"
// ,"org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
// configuration.addResource("classpath:/hadoop/core-site.xml");
// configuration.addResource("classpath:/hadoop/hdfs-site.xml");
// configuration.addResource("classpath:/hadoop/mapred-site.xml");
val fs = FileSystem.get(configuration)
val fdis: FSDataInputStream = fs.open(new Path(filePath))