ICode9
精准搜索请尝试： 精确搜索
首页
编程语言>
数据库
系统相关
互联网
其他分享
Python
Java
JavaScript
android
PHP
首页 > 其他分享> 文章详细

Flink算子和入门案例（wordcount）

2021-11-23 22:32:29  阅读：124  来源： 互联网

标签：Flink String val flink wordcount env 算子 apache org


文章目录

1.Flink入门案例wordcount

先导入pom依赖

 <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <flink.version>1.11.2</flink.version>
        <scala.binary.version>2.11</scala.binary.version>
        <scala.version>2.11.12</scala.version>
        <log4j.version>2.12.1</log4j.version>
    </properties>

    <dependencies>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-walkthrough-common_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>


        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>${log4j.version}</version>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.36</version>
        </dependency>


    </dependencies>


    <build>

        <plugins>


            <!-- Java Compiler -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>

            <!-- Scala Compiler -->
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <args>
                        <arg>-nobootcp</arg>
                    </args>
                </configuration>
            </plugin>


        </plugins>

    </build>

package com.liu.core

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

/**
 * @ Author : ld
 * @ Description : 实时统计word个数
 * @ Date : 2021/11/23 18:57
 * @ Version : 1.0
 */
object FlinkWordCount {
  def main(args: Array[String]): Unit = {
    //创建flink的环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
    env.setParallelism(2)
    //读取socket数据
    //启动master开启nc,没有的执行yum -install nc安装
    //nc -lk 8888
    env.socketTextStream("master",8888)
    //把单词拆分
      .flatMap(_.split(","))
    //转换成kv格式
      .map((_,1))
    //按单词分组
      .keyBy(_._1)
    //统计单词数量
      .sum(1)
    //打印结果
      .print()

    //启动flink
    env.execute()
  }
}

在这里插入图片描述
在这里插入图片描述

2.基于本地构建DataStream,基于文件构建DataStream，基于socket构建DataStream，自定义source

package com.liu.source

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


/**
 * @ Author : ld
 * @ Description : 
 * @ Date : 2021/11/23 19:26
 * @ Version : 1.0
 */
object Demo1Source {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    /**
     * 基于本地构建DataStream -- 有界流
     */
    val lisrDS: DataStream[Int] = env.fromCollection(List(1, 2, 3, 4, 5, 6, 7, 8, 9))
    lisrDS.print()

    /**
     * 基于文件构建DataStream --有界流
     */
    val studentDS: DataStream[String] = env.readTextFile("Flink/data/student.txt")

    studentDS
      .map(stu=>(stu.split(",")(4),1))
      .keyBy(_._1)
      .sum(1)
      .print()

    /**
     * 基于socket构建DataStream-- 无界流
     */
//    env.socketTextStream("master11",8888)
//      .print()


    /**
     * 自定义socket，实现SourceFunction接口
     */
    env.addSource(new MySource).print()

    env.execute()
  }
}

/**
 * 自定义source，实现SourceFunction接口
 * 实现run方法
 */
class MySource extends SourceFunction[Int]{
  /**
   * run方法只执行一次
   * @ param ctx:用于发送数据到下游task
   */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {
    var i=0
    while(true){//死循环，看完发送到下游结果就关闭吧
      //把数据发送到下游
      ctx.collect(i)
      //休眠50毫秒
      Thread.sleep(50)
      i+=1
    }

  }

  /**
   * cancel()方法再任务取消时执行用于回收资源
   */

  override def cancel(): Unit = {}
}

3.使用自定义source去读取MySQL数据库数据

package com.liu.source

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, ResultSet}

/**
 * @ Author : ld
 * @ Description : 
 * @ Date : 2021/11/23 20:05
 * @ Version : 1.0
 */
object Demo2MysqlSource {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(2)
    //使用自定义source
    val mysqlDS: DataStream[(Int, String, Int, String, String)] = env.addSource(new MysqlSource)
    mysqlDS.print()

    env.execute()
  }
}

/**
 * 自定义读取mysql---有界流
 * SourceFunction -- 单一source,run方法只会执行一次
 * ParallelSourceFunction-- 并行的source，并行度决定source个数
 * RichSourceFunction -- 比sourceFunction多了open和close方法
 * RichParallelSourceFunction --结合上面两个方法
 */

class MysqlSource extends RichSourceFunction[(Int, String, Int, String, String)] {
  /**
   * open方法会在run方法之前执行
   * @ param ctx
   */
  var conn: Connection = _

  override def open(parameters: Configuration): Unit = {
    //加载驱动
    Class.forName("com.mysql.jdbc.Driver")
    //建立连接
    conn = DriverManager.getConnection("jdbc:mysql://master:3306/test", "root", "123456")
  }
  /**
   * 在run方法后执行
   */
  override def close(): Unit = {
    //关闭连接
    conn.close()
  }


  override def run(ctx: SourceFunction.SourceContext[(Int, String, Int, String, String)]): Unit = {

    //查看数据
    val stat = conn.prepareStatement("select * from student")
    val res: ResultSet = stat.executeQuery()
    //解析数据
    while (res.next()) {
      val id: Int = res.getInt("id")
      val name: String = res.getString("name")
      val age: Int = res.getInt("age")
      val gender: String = res.getString("gender")
      val clazz: String = res.getString("clazz")
      //数据发送到下游
      ctx.collect((id, name, age, gender, clazz))

    }
  }


  override def cancel(): Unit = {

  }
}

4.map

package com.liu.transformation

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala._

/**
 * @ Author : ld
 * @ Description : 
 * @ Date : 2021/11/23 20:54
 * @ Version : 1.0
 */
object Demo1Map {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val lineDS: DataStream[String] = env.socketTextStream("master11", 8888)

    /**
     * map函数
     * 传入一个函数
     * 传入一个接口的实现类 --MapFunction
     */
    lineDS.map(new MapFunction[String,String]{
      override def map(t: String): String = {
      t +"ok"
      }
    }).print()

    env.execute()

  }
}

在这里插入图片描述

5.flatMap

package com.liu.transformation

import org.apache.flink.api.common.functions.{FlatMapFunction, RichFlatMapFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
/**
 * @ Author : ld
 * @ Description : 
 * @ Date : 2021/11/23 21:03
 * @ Version : 1.0
 */
object Demo2FlatMap {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(4)//设置并行度为4
    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * FlatMapFunction
     * RichSourceFunction--多了open和close方法，可以做初始化操作
     */

    val flatMapDS: DataStream[String] = linesDS.flatMap(new RichFlatMapFunction[String, String] {
      override def flatMap(line: String, out: Collector[String]): Unit = {
        /**
         * flatMap函数，每一条数据执行一次
         *
         * @ param line ： 一行数据
         * @ param out  ; 用于将数据发送到下游
         */
        line
          .split(",")
          .foreach(out.collect) //下面释内容简写
        //          .foreach(word=>{
        //            //发送数据
        //            out.collect(word)
        //          })
      }
    })

    flatMapDS.print()

    env.execute()
  }
}

在这里插入图片描述

6.filter

package com.liu.transformation

import org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.streaming.api.scala._
/**
 * @ Author : ld
 * @ Description : 
 * @ Date : 2021/11/23 21:17
 * @ Version : 1.0
 */
object Demo3Filter {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
     * filter的算子不是懒执行
     */
    val studentDS: DataStream[String] = env.readTextFile("Flink/data/student.txt")

    studentDS.filter(new FilterFunction[String]{
      override def filter(stu: String): Boolean = {
        //过滤出性别为男的所有学生
        stu.split(",")(3)=="男"
      }
    }).print()

    env.execute()
  }
}

在这里插入图片描述

7.KeyBy

package com.liu.transformation

import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.streaming.api.scala._
/**
 * @ Author : ld
 * @ Description : 
 * @ Date : 2021/11/23 21:25
 * @ Version : 1.0
 */
object Demo4KeyBy {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(3)

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

    /**
     * keyBy把相同的key发送到同一个task中
     */
    linesDS.keyBy(new KeySelector[String,String] {
      override def getKey(line: String): String ={
        line
      }
    }).print()

    env.execute()


  }
}

在这里插入图片描述

8.Reduce

package com.liu.transformation

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.scala._

/**
 * @ Author : ld
 * @ Description : 
 * @ Date : 2021/11/23 22:01
 * @ Version : 1.0
 */
object Demo5Reduce {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val lineDS: DataStream[String] = env.socketTextStream("master11", 8888)

    val keyByDS: KeyedStream[(String, Int), String] = lineDS
      .flatMap(_.split(","))
      .map((_, 1))
      .keyBy(_._1)

    /**
     * reduce:在keyBy之后进行聚合
     */
    keyByDS.reduce(new ReduceFunction[(String,Int)]{
      override def reduce(t: (String,Int), t1: (String,Int)): (String,Int) = {
        (t._1,t1._2+t1._2)
      }
    }).print()

    env.execute()
  }
}

9.Agg

package com.liu.transformation

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

/**
 * @ Author : ld
 * @ Description : 
 * @ Date : 2021/11/23 22:09
 * @ Version : 1.0
 */
object Demo6Agg {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val studentDS: DataStream[String] = env.readTextFile("Flink/data/student.txt")

    var stuDS: DataStream[Student] = studentDS.map(line => {
      val split = line.split(",")
      Student(split(0), split(1), split(2).toInt, split(3), split(4))
    })

    stuDS.keyBy(_.clazz)
      .sum("age")
      .print()
    /**
     * max 和 maxBy 之间的区别在于 max 返回流中的最大值，但 maxBy 返回具有最大值的键，
     */
    stuDS.keyBy(_.clazz)
      .maxBy("age")
      .print()

    env.execute()
  }
  case class Student(id:String,name:String,age:Int,gender:String,clazz:String)
}

10.Window

package com.liu.transformation

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
/**
 * @ Author : ld
 * @ Description : 
 * @ Date : 2021/11/23 21:52
 * @ Version : 1.0
 */
object Demo7Window  {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val linesDS: DataStream[String] = env.socketTextStream("master11", 8888)

    /**
     * 每5秒统计一次单词数量
     */
    linesDS
      .flatMap(_.split(","))
      .map((_,1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      .sum(1)
      .print()

    env.execute()



  }
}

标签：Flink,String,val,flink,wordcount,env,算子,apache,org
来源： https://blog.csdn.net/qq_43278189/article/details/121500550

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

关于我们 | 联系我们 | 留言反馈

专注分享技术，共同学习，共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有