java - 使用 Apache Flink 从 Web 获取 JSON 元素

Question

在阅读了 Apache Flink 的几个文档页面（官方文档、dataartisans）以及官方存储库中提供的示例之后，我不断看到它们用作流式传输已下载文件的数据源的示例，始终连接到本地主机。

我正在尝试使用 Apache Flink 下载包含动态数据的 JSON 文件。我的目的是尝试建立可以访问 JSON 文件作为 Apache Flink 输入源的 url，而不是使用另一个系统下载它并使用 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");