Flink DataStreamAPI與DataSetAPI應用案例實戰-Flink牛刀小試

摘要： 版權宣告：本套技術專欄是作者（秦凱新）平時工作的總結和昇華，通過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和叢集環境容量規劃等內容，請持續關注本套部落格。期待加入IOT時代最具戰鬥力的團隊。QQ郵箱地址：[email protected]，如有任何學術交流，可隨時聯絡。 ...

版權宣告：本套技術專欄是作者（秦凱新）平時工作的總結和昇華，通過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和叢集環境容量規劃等內容，請持續關注本套部落格。期待加入IOT時代最具戰鬥力的團隊。QQ郵箱地址：[email protected]，如有任何學術交流，可隨時聯絡。

1DataStreamAPI

1.1 DataStream Data Sources

• source是程式的資料來源輸入，你可以通過StreamExecutionEnvironment.addSource(sourceFunction)來為你的程式新增一個source。

• flink提供了大量的已經實現好的source方法，可以自定義source 通過實現sourceFunction介面來自定義無並行度的source。

  1 使用並行度為1的source
  public class MyNoParalleSource implements SourceFunction<Long>{
  
  private long count = 1L;
  
  private boolean isRunning = true;
  
  /**
  * 主要的方法
  * 啟動一個source
  * 大部分情況下，都需要在這個run方法中實現一個迴圈，這樣就可以迴圈產生資料了
  *
  * @param ctx
  * @throws Exception
  */
  @Override
  public void run(SourceContext<Long> ctx) throws Exception {
  while(isRunning){
  ctx.collect(count);
  count++;
  //每秒產生一條資料
  Thread.sleep(1000);
  }
  }
  * 取消一個cancel的時候會呼叫的方法
  @Override
  public void cancel() {
  isRunning = false;
  }
  }
  
  2 Main方法執行
  public class StreamingDemoWithMyNoPralalleSource {
  public static void main(String[] args) throws Exception {
  //獲取Flink的執行環境
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  //獲取資料來源
  DataStreamSource<Long> text = env.addSource(new MyNoParalleSource()).setParallelism(1); //注意：針對此source，並行度只能設定為1
  DataStream<Long> num = text.map(new MapFunction<Long, Long>() {
  @Override
  public Long map(Long value) throws Exception {
  System.out.println("接收到資料：" + value);
  return value;
  }
  });
  
  //每2秒鐘處理一次資料
  DataStream<Long> sum = num.timeWindowAll(Time.seconds(2)).sum(0);
  //列印結果
  sum.print().setParallelism(1);
  String jobName = StreamingDemoWithMyNoPralalleSource.class.getSimpleName();
  env.execute(jobName);
  }
  }
  複製程式碼

• 可以通過實現ParallelSourceFunction介面或者繼承RichParallelSourceFunction來自定義有並行度的source。繼承RichParallelSourceFunction的那些SourceFunction意味著它們都是並行執行的並且可能有一些資源需要open/close

  public class MyParalleSource implements ParallelSourceFunction<Long> {
  private long count = 1L;
  private boolean isRunning = true;
  
  /**
  * 主要的方法
  * 啟動一個source
  * 大部分情況下，都需要在這個run方法中實現一個迴圈，這樣就可以迴圈產生資料了
  *
  * @param ctx
  * @throws Exception
  */
  @Override
  public void run(SourceContext<Long> ctx) throws Exception {
  while(isRunning){
  ctx.collect(count);
  count++;
  //每秒產生一條資料
  Thread.sleep(1000);
  }
  }
  /**
  * 取消一個cancel的時候會呼叫的方法
  *
  */
  @Override
  public void cancel() {
  isRunning = false;
  }
  }
  
  public class StreamingDemoWithMyPralalleSource {
  public static void main(String[] args) throws Exception {
  //獲取Flink的執行環境
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
  //獲取資料來源
  DataStreamSource<Long> text = env.addSource(new MyParalleSource()).setParallelism(2);
  
  DataStream<Long> num = text.map(new MapFunction<Long, Long>() {
  @Override
  public Long map(Long value) throws Exception {
  System.out.println("接收到資料：" + value);
  return value;
  }
  });
  
  //每2秒鐘處理一次資料
  DataStream<Long> sum = num.timeWindowAll(Time.seconds(2)).sum(0);
  
  //列印結果
  sum.print().setParallelism(1);
  
  String jobName = StreamingDemoWithMyPralalleSource.class.getSimpleName();
  env.execute(jobName);
  }
  }
  
  
  public class MyRichParalleSource extends RichParallelSourceFunction<Long> {
  private long count = 1L;
  private boolean isRunning = true;
  /**
  * 主要的方法
  * 啟動一個source
  * 大部分情況下，都需要在這個run方法中實現一個迴圈，這樣就可以迴圈產生資料了
  * @param ctx
  * @throws Exception
  */
  @Override
  public void run(SourceContext<Long> ctx) throws Exception {
  while(isRunning){
  ctx.collect(count);
  count++;
  //每秒產生一條資料
  Thread.sleep(1000);
  }
  }
  /**
  * 取消一個cancel的時候會呼叫的方法
  *
  */
  @Override
  public void cancel() {
  isRunning = false;
  }
  /**
  * 這個方法只會在最開始的時候被呼叫一次
  * 實現獲取連結的程式碼
  * @param parameters
  * @throws Exception
  */
  @Override
  public void open(Configuration parameters) throws Exception {
  System.out.println("open.............");
  super.open(parameters);
  }
  
  /**
  * 實現關閉連結的程式碼
  * @throws Exception
  */
  @Override
  public void close() throws Exception {
  super.close();
  }
  }
  
  public class StreamingDemoWithMyRichPralalleSource {
  
  public static void main(String[] args) throws Exception {
  //獲取Flink的執行環境
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
  //獲取資料來源
  DataStreamSource<Long> text = env.addSource(new MyRichParalleSource()).setParallelism(2);
  
  DataStream<Long> num = text.map(new MapFunction<Long, Long>() {
  @Override
  public Long map(Long value) throws Exception {
  System.out.println("接收到資料：" + value);
  return value;
  }
  });
  
  //每2秒鐘處理一次資料
  DataStream<Long> sum = num.timeWindowAll(Time.seconds(2)).sum(0);
  
  //列印結果
  sum.print().setParallelism(1);
  
  String jobName = StreamingDemoWithMyRichPralalleSource.class.getSimpleName();
  env.execute(jobName);
  }
  }
  複製程式碼

• 基於檔案 readTextFile(path) 讀取文字檔案，檔案遵循TextInputFormat 讀取規則，逐行讀取並返回。

• 基於socket socketTextStream從socker中讀取資料，元素可以通過一個分隔符切開。

  public class SocketDemoFullCount {
  public static void main(String[] args) throws Exception{
  //獲取需要的埠號
  int port;
  try {
  ParameterTool parameterTool = ParameterTool.fromArgs(args);
  port = parameterTool.getInt("port");
  }catch (Exception e){
  System.err.println("No port set. use default port 9010--java");
  port = 9010;
  }
  //獲取flink的執行環境
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  String hostname = "SparkMaster";
  String delimiter = "\n";
  //連線socket獲取輸入的資料
  DataStreamSource<String> text = env.socketTextStream(hostname, port, delimiter);
  
  DataStream<Tuple2<Integer,Integer>> intData = text.map(new MapFunction<String, Tuple2<Integer,Integer>>() {
  @Override
  public Tuple2<Integer,Integer> map(String value) throws Exception {
  return new Tuple2<>(1,Integer.parseInt(value));
  }
  });
  
  intData.keyBy(0)
  .timeWindow(Time.seconds(5))
  .process(new ProcessWindowFunction<Tuple2<Integer,Integer>, String, Tuple, TimeWindow>() {
  @Override
  public void process(Tuple key, Context context, Iterable<Tuple2<Integer, Integer>> elements, Collector<String> out) throws Exception {
  System.out.println("執行process");
  long count = 0;
  for(Tuple2<Integer,Integer> element: elements){
  count++;
  }
  out.collect("window:"+context.window()+",count:"+count);
  }
  }).print();
  //這一行程式碼一定要實現，否則程式不執行
  env.execute("Socket window count");
  }
  }
  複製程式碼

• 基於集合 fromCollection(Collection) 通過java 的collection集合建立一個數據流，集合中的所有元素必須是相同型別的。

  public class StreamingFromCollection {
  public static void main(String[] args) throws Exception {
  //獲取Flink的執行環境
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  ArrayList<Integer> data = new ArrayList<>();
  data.add(10);
  data.add(15);
  data.add(20);
  
  //指定資料來源
  DataStreamSource<Integer> collectionData = env.fromCollection(data);
  //通map對資料進行處理
  DataStream<Integer> num = collectionData.map(new MapFunction<Integer, Integer>() {
  @Override
  public Integer map(Integer value) throws Exception {
  return value + 1;
  }
  });
  //直接列印
  num.print().setParallelism(1);
  env.execute("StreamingFromCollection");
  }
  複製程式碼

  }

• 自定義輸入 addSource 可以實現讀取第三方資料來源的資料 系統內建提供了一批connectors，聯結器會提供對應的source支援【kafka】

1.2 DataStream Transformations

• map：輸入一個元素，然後返回一個元素，中間可以做一些清洗轉換等操作

• flatmap：輸入一個元素，可以返回零個，一個或者多個元素

• keyBy：根據指定的key進行分組，相同key的資料會進入同一個分割槽

  dataStream.keyBy("someKey") // 指定物件中的 "someKey"欄位作為分組key
  dataStream.keyBy(0) //指定Tuple中的第一個元素作為分組key
  
  注意：以下型別是無法作為key的
  1：一個實體類物件，沒有重寫hashCode方法，並且依賴object的hashCode方法
  2：一個任意形式的陣列型別
  3：基本資料型別，int，long
  複製程式碼

• filter：過濾函式，對傳入的資料進行判斷，符合條件的資料會被留下。

  public class StreamingDemoFilter {
  public static void main(String[] args) throws Exception {
  //獲取Flink的執行環境
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  //獲取資料來源
  DataStreamSource<Long> text = env.addSource(new MyNoParalleSource()).setParallelism(1);//注意：針對此source，並行度只能設定為1
  DataStream<Long> num = text.map(new MapFunction<Long, Long>() {
  @Override
  public Long map(Long value) throws Exception {
  System.out.println("原始接收到資料：" + value);
  return value;
  }
  });
  
  //執行filter過濾，滿足條件的資料會被留下
  DataStream<Long> filterData = num.filter(new FilterFunction<Long>() {
  //把所有的奇數過濾掉
  @Override
  public boolean filter(Long value) throws Exception {
  return value % 2 == 0;
  }
  });
  
  DataStream<Long> resultData = filterData.map(new MapFunction<Long, Long>() {
  @Override
  public Long map(Long value) throws Exception {
  System.out.println("過濾之後的資料：" + value);
  return value;
  }
  });
  
  
  //每2秒鐘處理一次資料
  DataStream<Long> sum = resultData.timeWindowAll(Time.seconds(2)).sum(0);
  
  //列印結果
  sum.print().setParallelism(1);
  
  String jobName = StreamingDemoFilter.class.getSimpleName();
  env.execute(jobName);
  }
  }
  複製程式碼

• reduce：對資料進行聚合操作，結合當前元素和上一次reduce返回的值進行聚合操作，然後返回一個新的值

• aggregations：sum(),min(),max()等

• window：在後面單獨詳解

• Union：合併多個流，新的流會包含所有流中的資料，但是union是一個限制，就是所有合併的流型別必須是一致的。

  public class StreamingDemoUnion {
  public static void main(String[] args) throws Exception {
  //獲取Flink的執行環境
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
  //獲取資料來源
  DataStreamSource<Long> text1 = env.addSource(new MyNoParalleSource()).setParallelism(1);//注意：針對此source，並行度只能設定為1
  
  DataStreamSource<Long> text2 = env.addSource(new MyNoParalleSource()).setParallelism(1);
  
  //把text1和text2組裝到一起
  DataStream<Long> text = text1.union(text2);
  
  DataStream<Long> num = text.map(new MapFunction<Long, Long>() {
  @Override
  public Long map(Long value) throws Exception {
  System.out.println("原始接收到資料：" + value);
  return value;
  }
  });
  //每2秒鐘處理一次資料
  DataStream<Long> sum = num.timeWindowAll(Time.seconds(2)).sum(0);
  //列印結果
  sum.print().setParallelism(1);
  String jobName = StreamingDemoUnion.class.getSimpleName();
  env.execute(jobName);
  }
  複製程式碼

  }

• Connect：和union類似，但是隻能連線兩個流，兩個流的資料型別可以不同，會對兩個流中的資料應用不同的處理方法。

• CoMap, CoFlatMap：在ConnectedStreams中需要使用這種函式，類似於map和flatmap

  public class StreamingDemoConnect {
  
  public static void main(String[] args) throws Exception {
  //獲取Flink的執行環境
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  //獲取資料來源
  DataStreamSource<Long> text1 = env.addSource(new MyNoParalleSource()).setParallelism(1);//注意：針對此source，並行度只能設定為1
  DataStreamSource<Long> text2 = env.addSource(new MyNoParalleSource()).setParallelism(1);
  SingleOutputStreamOperator<String> text2_str = text2.map(new MapFunction<Long, String>() {
  @Override
  public String map(Long value) throws Exception {
  return "str_" + value;
  }
  });
  
  ConnectedStreams<Long, String> connectStream = text1.connect(text2_str);
  
  SingleOutputStreamOperator<Object> result = connectStream.map(new CoMapFunction<Long, String, Object>() {
  @Override
  public Object map1(Long value) throws Exception {
  return value;
  }
  @Override
  public Object map2(String value) throws Exception {
  return value;
  }
  });
  //列印結果
  result.print().setParallelism(1);
  String jobName = StreamingDemoConnect.class.getSimpleName();
  env.execute(jobName);
  }
  }
  複製程式碼

• Split：根據規則把一個數據流切分為多個流：

• Select：和split配合使用，選擇切分後的流

  public class StreamingDemoSplit {
  public static void main(String[] args) throws Exception {
  //獲取Flink的執行環境
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
  //獲取資料來源
  DataStreamSource<Long> text = env.addSource(new MyNoParalleSource()).setParallelism(1);//注意：針對此source，並行度只能設定為1
  
  //對流進行切分，按照資料的奇偶性進行區分
  SplitStream<Long> splitStream = text.split(new OutputSelector<Long>() {
  @Override
  public Iterable<String> select(Long value) {
  ArrayList<String> outPut = new ArrayList<>();
  if (value % 2 == 0) {
  outPut.add("even");//偶數
  } else {
  outPut.add("odd");//奇數
  }
  return outPut;
  }
  });
  
  //選擇一個或者多個切分後的流
  DataStream<Long> evenStream = splitStream.select("even");
  DataStream<Long> oddStream = splitStream.select("odd");
  
  DataStream<Long> moreStream = splitStream.select("odd","even");
  
  
  //列印結果
  moreStream.print().setParallelism(1);
  
  String jobName = StreamingDemoSplit.class.getSimpleName();
  env.execute(jobName);
  }
  }
  複製程式碼