amazon-web-services - 在spark MapTask中调用http请求和读取输入流的有效方法是什么

Question

请看下面的代码示例

JavaRDD<String> mapRDD = filteredRecords
            .map(new Function<String, String>() {

                @Override
                public String call(String url) throws Exception {
                    BufferedReader in = null;
                    URL formatURL = new URL((url.replaceAll("\"", ""))
                            .trim());
                    try {
                        HttpURLConnection con = (HttpURLConnection) formatURL
                                .openConnection();
                        in = new BufferedReader(new InputStreamReader(con
                                .getInputStream()));

                        return in.readLine();
                    } finally {
                        if (in != null) {
                            in.close();
                        }
                    }
                }
            }); 

这里的url是http GET请求。例子

http://ip:port/cyb/test?event=movie&id=604568837&name=SID&timestamp_secs=1460494800&timestamp_millis=1461729600000&back_up_id=676700166

这段代码很慢。 IP 和端口是随机的，负载是分布式的，所以 ip 可以有 20 个不同的端口值，所以我看不到瓶颈。

当我发表评论时

 in = new BufferedReader(new InputStreamReader(con
                            .getInputStream()));

                    return in.readLine();

代码太快了。注意:要处理的输入数据为 10GB。使用 spark 从 S3 读取数据。

我在使用 BufferedReader 或 InputStreamReader 时有什么问题吗？我不能在 spark 中使用 foreach，因为我必须从服务器获取响应并且需要将 JAVARdd 保存为 HDFS 上的文本文件。

如果我们使用如下的 mappartition 代码

JavaRDD<String> mapRDD = filteredRecords.mapPartitions(new FlatMapFunction<Iterator<String>, String>() {

        @Override
        public Iterable<String> call(Iterator<String> tuple) throws Exception {

            final List<String> rddList = new ArrayList<String>();
            Iterable<String> iterable = new Iterable<String>() {

                @Override
                public Iterator<String> iterator() {
                    return rddList.iterator();
                }
            };
            while(tuple.hasNext()) {
                URL formatURL = new URL((tuple.next().replaceAll("\"", ""))
                        .trim());
                HttpURLConnection con = (HttpURLConnection) formatURL
                        .openConnection();
                try(BufferedReader br = new BufferedReader(new InputStreamReader(con
                        .getInputStream()))) {

                    rddList.add(br.readLine());

                } catch (IOException ex) {
                    return rddList;
                }
            }
            return iterable;
        }
    }); 

对于每条记录，我们也在做同样的事情......不是吗？

Answer 1

您目前正在使用

map function

为分区中的每一行创建一个 url 请求。

你可以使用

mapPartition

这将使代码运行得更快，因为它只创建一次与服务器的连接，即每个分区只有一个连接。