Flink 快速入门开发 ( 基于scala )

pom 中导入依赖

  1.     <properties>
  2.         <maven.compiler.source>1.8</maven.compiler.source>
  3.         <maven.compiler.target>1.8</maven.compiler.target>
  4.         <encoding>UTF-8</encoding>
  5.         <scala.version>2.11.8</scala.version>
  6.         <scala.binary.version>2.11</scala.binary.version>
  7.         <hadoop.version>2.7.6</hadoop.version>
  8.         <flink.version>1.6.1</flink.version>
  9.         <mysql.version>5.1.48</mysql.version>
  10.         <fastjson.version>1.2.51</fastjson.version>
  11.     </properties>
  13.     <dependencies>
  14.         <dependency>
  15.             <groupId>org.scala-lang</groupId>
  16.             <artifactId>scala-library</artifactId>
  17.             <version>${scala.version}</version>
  18.         </dependency>
  19.         <dependency>
  20.             <groupId>org.apache.flink</groupId>
  21.             <artifactId>flink-java</artifactId>
  22.             <version>${flink.version}</version>
  23.         </dependency>
  24.         <dependency>
  25.             <groupId>org.apache.flink</groupId>
  26.             <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
  27.             <version>${flink.version}</version>
  28.         </dependency>
  29.         <dependency>
  30.             <groupId>org.apache.flink</groupId>
  31.             <artifactId>flink-scala_${scala.binary.version}</artifactId>
  32.             <version>${flink.version}</version>
  33.         </dependency>
  34.         <dependency>
  35.             <groupId>org.apache.flink</groupId>
  36.             <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
  37.             <version>${flink.version}</version>
  38.         </dependency>
  39.         <dependency>
  40.             <groupId>org.apache.flink</groupId>
  41.             <artifactId>flink-table_${scala.binary.version}</artifactId>
  42.             <version>${flink.version}</version>
  43.         </dependency>
  44.         <dependency>
  45.             <groupId>org.apache.flink</groupId>
  46.             <artifactId>flink-clients_${scala.binary.version}</artifactId>
  47.             <version>${flink.version}</version>
  48.         </dependency>
  49.         <dependency>
  50.             <groupId>org.apache.flink</groupId>
  51.             <artifactId>flink-connector-kafka-0.10_${scala.binary.version}</artifactId>
  52.             <version>${flink.version}</version>
  53.         </dependency>
  54.         <dependency>
  55.             <groupId>org.apache.hadoop</groupId>
  56.             <artifactId>hadoop-client</artifactId>
  57.             <version>${hadoop.version}</version>
  58.         </dependency>
  60.         <dependency>
  61.             <groupId>mysql</groupId>
  62.             <artifactId>mysql-connector-java</artifactId>
  63.             <version>${mysql.version}</version>
  64.         </dependency>
  65.         <dependency>
  66.             <groupId>com.alibaba</groupId>
  67.             <artifactId>fastjson</artifactId>
  68.             <version>${fastjson.version}</version>
  69.         </dependency>
  70.     </dependencies>
  72.     <build>
  73.         <sourceDirectory>src/main/scala</sourceDirectory>
  74.         <testSourceDirectory>src/test/scala</testSourceDirectory>
  75.         <plugins>
  76.             <plugin>
  77.                 <groupId>net.alchim31.maven</groupId>
  78.                 <artifactId>scala-maven-plugin</artifactId>
  79.                 <version>3.2.0</version>
  80.                 <executions>
  81.                     <execution>
  82.                         <goals>
  83.                             <goal>compile</goal>
  84.                             <goal>testCompile</goal>
  85.                         </goals>
  86.                         <configuration>
  87.                             <args>
  88.                                 <!-- <arg>-make:transitive</arg> -->
  89.                                 <arg>-dependencyfile</arg>
  90.                                 <arg>${project.build.directory}/.scala_dependencies</arg>
  91.                             </args>
  92.                         </configuration>
  93.                     </execution>
  94.                 </executions>
  95.             </plugin>
  96.             <plugin>
  97.                 <groupId>org.apache.maven.plugins</groupId>
  98.                 <artifactId>maven-surefire-plugin</artifactId>
  99.                 <version>2.18.1</version>
  100.                 <configuration>
  101.                     <useFile>false</useFile>
  102.                     <disableXmlReport>true</disableXmlReport>
  103.                     <includes>
  104.                         <include>**/*Test.*</include>
  105.                         <include>**/*Suite.*</include>
  106.                     </includes>
  107.                 </configuration>
  108.             </plugin>
  109.             <plugin>
  110.                 <groupId>org.apache.maven.plugins</groupId>
  111.                 <artifactId>maven-shade-plugin</artifactId>
  112.                 <version>3.2.1</version>
  113.                 <executions>
  114.                     <execution>
  115.                         <phase>package</phase>
  116.                         <goals>
  117.                             <goal>shade</goal>
  118.                         </goals>
  119.                         <configuration>
  120.                             <filters>
  121.                                 <filter>
  122.                                     <artifact>*:*</artifact>
  123.                                     <excludes>
  124.                                         <exclude>META-INF/*.SF</exclude>
  125.                                         <exclude>META-INF/*.DSA</exclude>
  126.                                         <exclude>META-INF/*.RSA</exclude>
  127.                                     </excludes>
  128.                                 </filter>
  129.                             </filters>
  130.                             <transformers>
  131.                                 <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  132.                                     <mainClass>org.apache.spark.WordCount</mainClass>
  133.                                 </transformer>
  134.                             </transformers>
  135.                         </configuration>
  136.                     </execution>
  137.                 </executions>
  138.             </plugin>
  139.         </plugins>
  140.     </build>

批处理 （ 文件 —> world count ）

  1. import org.apache.flink.api.scala.ExecutionEnvironment
  2. import org.apache.flink.api.scala._
  4. /**
  5.   * @author yanglibin
  6.   * @create 2020-02-28 11:17
  7.   */
  8. object WorldCount {
  9.   def wc1(filePath :String) = {
  10.     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
  12.     env.readTextFile(filePath)
  13.       .flatMap(_.split(" "))
  14.       .map( (_,1) )
  15.       .groupBy(0)
  16.       .sum(1)
  17.       .print()
  18.   }
  19. }

常见问题

  1. Error:(15, 15) could not find implicit value for evidence parameter of type org.apache.flink.api.common.typeinfo.TypeInformation[String]
  2.       .flatMap(_.split(" "))
  4. 解决:
  5. import org.apache.flink.api.scala._

流处理

Flink 是一个流处理框架, 而且是一个事件驱动的框架

file

  1.   def wc2(filePath: String) = {
  2.     // 获取环境
  3.     val environment = StreamExecutionEnvironment.getExecutionEnvironment
  4.     // 设置 cpu 核数
  5.     environment.setParallelism(4)
  7.     environment.readTextFile(filePath)
  8.       .flatMap(_.split(" "))
  9.       .map( (_,1) )
  10.       .keyBy(0)
  11.       .sum(1)
  12.       .print()
  14.     // 执行
  15.     environment.execute()
  16.   }

socket

  1.   /**
  2.     * 无边界
  3.     * @param hostname
  4.     * @param port
  5.     * @return
  6.     */
  7.   def wc3(hostname:String, port: Int) = {
  8.     val environment = StreamExecutionEnvironment.getExecutionEnvironment
  10.     environment.setParallelism(4)
  11.     // 从 socket 中获取网络数据
  12.     environment.socketTextStream(hostname, port)
  13.       .flatMap(_.split(" "))
  14.       .map( (_,1) )
  15.       .keyBy(0)
  16.       .sum(1)
  17.       .print()
  19.     environment.execute()
  20.   }

Flink 集群环境搭建

Standalone模式

  1. # 下载地址: https://flink.apache.org/downloads.html
  2. # 下载时注意 scala 的版本
  3. tar -xvf /usr/local/src/flink-1.9.2-bin-scala_2.11.tgz  -C  /usr/local/
  4. ln -sv /usr/local/flink-1.9.2  /usr/local/flink
  6. ###     flink 配置该用     ###
  7. # conf/flink-conf.yaml
  8. jobmanager.rpc.address     flink-master
  10. # conf/master
  11. flink-master
  13. # conf/slaves
  14. flink-node-01
  15. flink-node-02
  19. ### flink cluster 启动  ###
  20. bin/start-cluster.sh

yarn 模式

1) Adding the Hadoop classpath to Flink

  1. #############     hadoop 环境    ##############
  2. # Check if deprecated HADOOP_HOME is set, and specify config path to HADOOP_CONF_DIR if it's empty.
  3. if [ -z "$HADOOP_CONF_DIR" ]; then
  4.     if [ -n "$HADOOP_HOME" ]; then
  5.         # HADOOP_HOME is set. Check if its a Hadoop 1.x or 2.x HADOOP_HOME path
  6.         if [ -d "$HADOOP_HOME/conf" ]; then
  7.             # its a Hadoop 1.x
  8.             HADOOP_CONF_DIR="$HADOOP_HOME/conf"
  9.         fi
  10.         if [ -d "$HADOOP_HOME/etc/hadoop" ]; then
  11.             # Its Hadoop 2.2+
  12.             HADOOP_CONF_DIR="$HADOOP_HOME/etc/hadoop"
  13.         fi
  14.     fi
  15. fi
  17. # try and set HADOOP_CONF_DIR to some common default if it's not set
  18. if [ -z "$HADOOP_CONF_DIR" ]; then
  19.     if [ -d "/etc/hadoop/conf" ]; then
  20.         echo "Setting HADOOP_CONF_DIR=/etc/hadoop/conf because no HADOOP_CONF_DIR was set."
  21.         HADOOP_CONF_DIR="/etc/hadoop/conf"
  22.     fi
  23. fi

2) Putting the required jar files into /lib directory of the Flink distribution

  1. # run Flink on YARN, connect to HDFS, connect to HBase, 
  2. # or use some Hadoop-based file system connector
  4. # 下载 对应的依赖lib 包 ( Pre-bundled Hadoop 2.7.5  )  到 flink lib 目录下
  5. wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.7.5-10.0/flink-shaded-hadoop-2-uber-2.7.5-10.0.jar -O /usr/local/flink/lib/flink-shaded-hadoop-2-uber-2.7.5-10.0.jar

flink job 运行

standalone 模式下

web UI 界面操作 （ 略 ）

命令行运行

  1. bin/flink run -c com.ylb.WorldCount /tmp/flink-demo.jar
  2. # -c : 指定 jar 包中要运行的类名

yarn 模式下

session-cluster

这里的session 表示的是 资源不释放，

使用:

  1. 先 yarn-session.sh 启动一个 session， 用来处理 flink
  2. 启动 flink 任务 
  1. # 1. 启动一个 session cluster test 实例
  2. bin/yarn-session.sh -n 2 -s 2 -jm 1024 -tm 1024 -nm test -d
  3. # 启动之后在本机（其他机器没有）的临时目录（/tmp 目录）下会生成一个文件：.yarn-properties-(用户名)
  5. # 2. 执行任务( 向上面创建的 yarn-application )
  6. bin/flink run -c com.ylb.WorldCount /tmp/flink-demo.jar
  8. # 3. kill yarn application 
  9. yarn application -kill application_1583028240987_0002  # 对应自己的app_id
  11. # 参数介绍
  12.      -n,--container <arg>   Number of YARN container to allocate (=Number of Task Managers)
  13.    Optional
  14.      -at,--applicationType <arg>     Set a custom application type for the application on YARN
  15.      -D <property=value>             use value for given property
  16.      -d,--detached                   If present, runs the job in detached mode
  17.      -h,--help                       Help for the Yarn session CLI.
  18.      -id,--applicationId <arg>       Attach to running YARN session
  19.      -j,--jar <arg>                  Path to Flink jar file
  20.      -jm,--jobManagerMemory <arg>    Memory for JobManager Container with optional unit (default: MB)
  21.      -m,--jobmanager <arg>           Address of the JobManager (master) to which to connect. Use this flag to connect to a different JobManager than the one specified in the configuration.
  22.      -n,--container <arg>            Number of YARN container to allocate (=Number of Task Managers)
  23.      -nl,--nodeLabel <arg>           Specify YARN node label for the YARN application
  24.      -nm,--name <arg>                Set a custom name for the application on YARN
  25.      -q,--query                      Display available YARN resources (memory, cores)
  26.      -qu,--queue <arg>               Specify YARN queue.
  27.      -s,--slots <arg>                Number of slots per TaskManager
  28.      -sae,--shutdownOnAttachedExit   If the job is submitted in attached mode, perform a best-effort cluster shutdown when the CLI is terminated abruptly, e.g., in response to a user interrupt, such
  29.                                      as typing Ctrl + C.
  30.      -st,--streaming                 Start Flink in streaming mode
  31.      -t,--ship <arg>                 Ship files in the specified directory (t for transfer)
  32.      -tm,--taskManagerMemory <arg>   Memory per TaskManager Container with optional unit (default: MB)
  33.      -yd,--yarndetached              If present, runs the job in detached mode (deprecated; use non-YARN specific option instead)
  34.      -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths for high availability mode

Per Job Cluster

每一个 flink job 申请一个独立的 yarn-application 

  1. bin/flink run -m yarn-cluster -c com.ylb.WorldCount /tmp/flink-demo.jar
  2. # 注意要删除 /tmp/.yarn-properties-xxxx

两者的区别:

资源是否长时间占用不释放

任务链: 将并行度相同的一对一关系方法形成一个完整的任务来执行。
禁用操作链条: env.disableOperatorChaining() 会导致操作不会链接在一起形成完整的任务， 一个subTask 就是一个Task
startNewChain() : 从当前方法开始产生新的任务链条