Как осуществлять потоковую загрузку с использованием Scalaj-Http и Hadoop HttpFs - PullRequest
Новая
       103

Как осуществлять потоковую загрузку с использованием Scalaj-Http и Hadoop HttpFs

0 голосов
/ 01 сентября 2018

Мой вопрос заключается в том, как использовать поток с буферизацией при использовании Scalaj-Http.

Я написал следующий код, который является полным рабочим примером, который загружает файл из Hadoop HDFS с использованием HttpFS. Моя цель - обрабатывать очень большие файлы, и для этого потребуется буферизованный подход с несколькими записями ввода-вывода в локальный файл.

Мне не удалось найти документацию о том, как использовать поток с интерфейсом ScalaJ-Http. Мне интересен пример загрузки и выгрузки, который может обрабатывать большие файлы размером несколько ГБ. Мой код ниже использует буферизацию памяти, которая подходит только для прототипирования. 

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
    }
}
...