如何從Zeppelin加載Flink流數據

我開始在Zeppelin上使用Flink並嘗試運行流中最簡單的程序：wordcount。當我使用終端在本地模式下運行此代碼時，它工作。如何從Zeppelin加載Flink流數據

這是我如何做到這一點：https://ci.apache.org/projects/flink/flink-docs-release-1.2/quickstart/setup_quickstart.html

這是代碼：

object SocketWindowWordCount { 

    /** Main program method */ 
    def main(args: Array[String]) : Unit = { 

    // the host and the port to connect to 
    var hostname: String = "localhost" 
    var port: Int = 0 

    try { 
     val params = ParameterTool.fromArgs(args) 
     hostname = if (params.has("hostname")) params.get("hostname") else "localhost"va 
     port = params.getInt("port") 
    } catch { 
     case e: Exception => { 
     System.err.println("No port specified. Please run 'SocketWindowWordCount " + 
      "--hostname <hostname> --port <port>', where hostname (localhost by default) and port " + 
      "is the address of the text server") 
     System.err.println("To start a simple text server, run 'netcat -l <port>' " + 
      "and type the input text into the command line") 
     return 
     } 
    } 

    // get the execution environment 
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment 

    // get input data by connecting to the socket 
    val text = env.socketTextStream(hostname, port, '\n') 

    // parse the data, group it, window it, and aggregate the counts 
    val windowCounts = text 
      .flatMap { w => w.split("\\s") } 
      .map { w => WordWithCount(w, 1) } 
      .keyBy("word") 
      .timeWindow(Time.seconds(5)) 
      .sum("count") 

    // print the results with a single thread, rather than in parallel 
    windowCounts.print().setParallelism(1) 

    env.execute("Socket Window WordCount") 
    } 

    /** Data type for words with count */ 
    case class WordWithCount(word: String, count: Long) 
}

然而，當我嘗試運行齊柏林這個代碼...我不如何做到這一點。我想我也需要在終端上打開一個套接字，但我不知道如何讓這個Flink連接到這個端口。我應該在Flink上打開幾個窗口嗎？

我可以摘要問題在：如何從Zeppelin加載flink流數據？

我需要在最後添加一些Scala的方法，如：事先

import scala.io.Source 

val programRunning = Source.from"Socket"...?

感謝您的幫助！ :)

來源

2017-05-26 Borja

嗨，我不是一個flink用戶，所以不知道如何正確使用flink流，但這可能對您有所幫助。 [Flink使用Zeppelin，2016.09]（https://rawkintrevo.org/2016/09/30/big-data-for-n00bs-my-first-streaming-flink-program-twitter/） – 1ambda

在終端，你可以使用

nc -lk 9999

建立綁定到端口9999，將發送不管它接收標準輸入到客戶的服務。

Apache Zeppelin: A friendlier way to Flink也可能有幫助。

來源

2017-05-30 14:35:11

**非常感謝**文檔非常有用，解決了我的問題！ – Borja

如何從Zeppelin加載Flink流數據

回答

相關問題