Spark跟Flink的区别、常见的Source

package com.shujia.core

import org.apache.flink.streaming.api.scala._

object Demo01WordCount {
  def main(args: Array[String]): Unit = {
    
    val env: StreamExecutionEnvironment= StreamExecutionEnvironment.getExecutionEnvironment

    //默认并行度等于CPU的逻辑核数 相当于是任务的一个并行度
    env.setParallelism(2)

    
    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    //对每一条数据进行切分
    val wordsDS: DataStream[String] = linesDS
      .flatMap(line => line.split(","))

    //将每个单词变成 K V 格式
    val wordsKVDS: DataStream[(String, Int)] = wordsDS.map(word => (word, 1))

    //按照每个单词进行分组
    val keyByDS: KeyedStream[(String, Int), String] = wordsKVDS.keyBy(kv => kv._1)

    //统计每个单词的数量 指定位置进行累加
    val wordCntDS: DataStream[(String, Int)] = keyByDS.sum(1)

    

    //将结果打印
    wordCntDS.print()

    //启动任务
    env.execute("Demo01WordCount")
  }
}

package com.shujia.source

import org.apache.flink.streaming.api.scala._

object Demo01ListSource {
  def main(args: Array[String]): Unit = {
    //创建Flink的入口
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //基于本地集合，例如List构建DataStream
    val linesList: List[String] = List("java,java,spark", "spark,spark,java")

    val linesDS: DataStream[String] = env.fromCollection(linesList)

    linesDS
      .flatMap(_.split(","))
      .map(word=>(word,1))
      .keyBy(kv=>kv._1)
      .sum(1)
      .print()

    //启动任务
    env.execute()

  }
}

package com.shujia.source

import org.apache.flink.streaming.api.scala._

object Demo02FileSource {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 将文件作为Source构建DataStream --> 有界流
    val stuDS: DataStream[String] = env.readTextFile("Flink/data/stu/students.txt")

    // 统计班级人数
    stuDS
      .map(line => (line.split(",")(4), 1))
      .keyBy(kv => kv._1)
      .sum(1)
      .print()

    env.execute()
  }
}

package com.shujia.source

import org.apache.flink.streaming.api.scala._

object Demo03SocketSource {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //以Socket作为Source构建DataStream
    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)
    
    
    linesDS
      .flatMap(line=>{
        print("flatMap")
        line.split(",")
      })
      .map(word=>(word,1))
      .keyBy(kv=>kv._1)
      .sum(1)
//      .print()

    env.execute()
  }
}

package com.shujia.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo04MySource {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //使用自定义的Source
    val myDS: DataStream[Int] = env.addSource(new MySource())
    myDS.print()

    env.execute()
  }
}

class MySource extends SourceFunction[Int] {
  
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {
    //每调用一次collect方法 即可获得一条数据
    var i:Int=0
    while(true){
      ctx.collect(i)
      i+=1
      Thread.sleep(3000)
    }
  }

  //当程序结束是会被调用 主要用于回收资源
  override def cancel(): Unit = {

  }
}