使用Apache Flink从Web获取JSON元素

Question

在阅读了Apache Flink(官方文档,数据公司)的几个文档页面以及官方存储库中提供的示例之后,我不断看到它们用作流式化已下载文件的数据源的示例,始终连接到localhost.

我正在尝试使用Apache Flink下载包含动态数据的JSON文件.我的目的是尝试建立我可以访问JSON文件的URL作为Apache Flink的输入源,而不是使用其他系统下载它并使用Apache Flink处理下载的文件.

是否有可能与Apache Flink建立这种网络连接？

Answer 1

您可以定义要下载的URL作为输入DataStream，然后从中下载文档MapFunction。以下代码演示了这一点：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<String> inputURLs = env.fromElements("http://www.json.org/index.html");

inputURLs.map(new MapFunction<String, String>() {
    @Override
    public String map(String s) throws Exception {
        URL url = new URL(s);
        InputStream is = url.openStream();

        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(is));

        StringBuilder builder = new StringBuilder();
        String line;

        try {
            while ((line = bufferedReader.readLine()) != null) {
                builder.append(line + "\n");
            }
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }

        try {
            bufferedReader.close();
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }

        return builder.toString();
    }
}).print();

env.execute("URL download job");