使用Apache Flink從網頁獲取JSON元素

Question

在閱讀Apache Flink的幾個文檔頁面（official documentation,dataartisans）以及official repository中提供的示例之後，我會繼續看到它們用作數據源來流式處理文件的示例已經下載，總是連接到本地主機。

我正在嘗試使用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");