Flink快速入門二 Source程式設計

3.dataset source程式設計

當前主流的(tez/spark/flink)計算都是通過構建dag圖，然後觸發執行的。flink 的dataset和datastream的source程式設計很是類似。只是呼叫的api不一樣而已。source程式設計官網指導

datastream api 是用於進行stream流計算開發的api。如下是列舉的幾種簡單的source程式設計api。

2.1讀取socket中資料生成stream

省略，請參flink快速入門一(簡介以及wc程式設計）

2.2讀取text檔案(夾)中的資料生成stream

def fromtextfile

(env: streamexecutionenvironment)
: unit =

此api非常的方便測試

def fromcollection

(env: streamexecutionenvironment)
: unit =

package com.wsk.flink.source

import org.apache.flink.streaming.api.scala.
import org.apache.flink.api.scala._
/** * datastream source 程式設計
*/def main
(args: array[string]
): unit =
/** *從 集合中生成 datastream
* @param env
*/def fromcollection
(env: streamexecutionenvironment)
: unit =
/** * 從text檔案（夾）中生成 datastream
** @param env
*/def fromtextfile
(env: streamexecutionenvironment)
: unit =
}

3.1讀取csv檔案中的資料生成dataset

def fromcsv

(env: executionenvironment)
: unit =

def fromrecursive

(env: executionenvironment)
: unit =

flink預設支援多種壓縮格式檔案的自動解壓讀取，但是要注意壓縮檔案本身不支援並行讀取和計算。

def fromcomcompressfile

(env: executionenvironment)
: unit =

package com.wsk.flink.source

import org.apache.flink.api.scala.executionenvironment
import org.apache.flink.api.scala._
import org.apache.flink.configuration.configuration
/** * batch source 程式設計
*/def main
(args: array[string]
): unit =
/** * 從 csv中獲取 dataset
** @param env
*/def fromcsv
(env: executionenvironment)
: unit =
/** * 從遞迴的資料夾中讀取資料
** @param env
*/def fromrecursive
(env: executionenvironment)
: unit =
/** * 讀取壓縮檔案
** @param env
*/def fromcomcompressfile
(env: executionenvironment)
: unit =
case
class
teacher
(name: string, age: int)
}

Flink 快速入門案例二（參考官方文件）

具有一定實際意義的流處理程式。結合信用卡欺詐驗證場景，實現的具體demo。package com.sanxiau import org.apache.flink.streaming.api.datastream.datastream import org.apache.flink.streaming...

Flink 基礎入門

作為致力於提供 stateful computations over data streams 的 下一代大資料處理引擎 flink 在流式計算領域帶來了諸多全新特性。本場 chat，我們將從 0 開始，幫助讀者入門 flink，主要內容包括 流式計算的目標與傳統架構 flink 基礎結構 flin...

二 快速入門

一 springboot預設埠號為8080 在沒有主啟動類的情況下 package com.dudu.service import org.springframework.boot.autoconfigure.enableautoconfiguration import org.springfram...