我的问题是在使用scalajhttp时如何使用缓冲流。
我已经编写了以下代码,这是一个完整的工作示例,将使用httpfs从hadoop hdfs下载一个文件。我的目标是处理非常大的文件,这将需要使用一种缓冲方法,将多个i/o写入本地文件。
我还没有找到关于如何使用scalajhttp接口的流的文档。我对一个下载和上传都可以处理大型多gb文件的例子很感兴趣。我下面的代码使用内存缓冲,这只适用于原型。
import scalaj.http._
import ujson.Js
import java.text.SimpleDateFormat
import java.net.SocketTimeoutException
import java.io.InputStream
import java.io.BufferedOutputStream
import java.io.FileOutputStream
import java.io.FileNotFoundException
object CopyFileFromHdfs {
def main(args: Array[String]) {
val host = "hadoop.example.com"
val user = "root"
var dstFile = ""
var srcFile = ""
val operation = "OPEN"
val port = 14000
System.setProperty("sun.net.http.allowRestrictedHeaders", "true")
if (args.length != 2)
{
println("Error: Missing or too many arguments")
println("Usage: CopyFileFromHdfs <srcfile> <dstfile>")
System.exit(1)
}
srcFile = args(0)
dstFile = args(1)
//********************************************************************************
// Create the URL string that we will use to connect to Hadoop HttpFS
//
// The string will look like this:
// http://root@123.456.789.012:14000/webhdfs/v1/?user.name=root&op=OPEN
//********************************************************************************
val url = makeHttpfsUrl(host, user, srcFile, operation, port)
//********************************************************************************
// Using HTTP, call the HttpFS server
//
// Exceptions:
// java.net.SocketTimeoutException
// java.net.UnknownHostException
// java.lang.IllegalArgumentException
// Remote Exceptions:
// java.io.FileNotFoundException
// com.sun.jersey.api.NotFoundException
//********************************************************************************
try {
var response = Http(url)
.timeout(connTimeoutMs = 1000, readTimeoutMs = 5000)
.asBytes
//********************************************************************************
// Check for an error. We are expecting an HTTP 200 response
//********************************************************************************
if (response.code < 200 || response.code > 299)
{
val data = ujson.read(response.body)
printf("Error: Cannot download file: %s\n", dstFile)
println(removeQuotes(data("RemoteException")("message").str))
println(removeQuotes(data("RemoteException")("exception").str))
System.exit(1)
}
val is = new FileOutputStream(dstFile)
val bs = new BufferedOutputStream(is)
bs.write(response.body, 0, response.body.length)
bs.close()
is.close()
} catch {
case e: SocketTimeoutException => {
printf("Error: Cannot connect to host %s on port %d\n", host, port)
println(e)
System.exit(1);
}
case e: Exception => {
printf("Error (other): Cannot download file %s\n", srcFile)
println(e)
System.exit(1);
}
}
printf("Success: File downloaded. %s -> %s\n", srcFile, dstFile)
System.exit(0)
}
//********************************************************************************
// The Json strings are surrounded by quotes.
// This function will remove them (only at the start and the end).
//********************************************************************************
def removeQuotes(str: String): String = {
// This expression will delete quotes at the beginning and end of a string
return str.replaceAll("^\"|\"$", "");
}
//********************************************************************************
// Create the URL string that we will use to connect to Hadoop HttpFS
//
// The string will look like this:
// http://root@123.456.789.012:14000/webhdfs/v1/?user.name=root&op=LISTSTATUS
//********************************************************************************
def makeHttpfsUrl(
host: String,
user: String,
hdfsPath: String,
operation: String,
port: Integer) : String = {
var url = "http://" + user + "@" + host + ":" + port.toString + "/webhdfs/v1"
if (hdfsPath(0) == '/')
url += hdfsPath
else
url += "/" + hdfsPath
url += "?user.name=" + user + "&op=" + operation
return url
}
}
暂无答案!
目前还没有任何答案,快来回答吧!