ToB企服应用市场:ToB评测及商务社交产业平台

标题: Flink入门实战详解 [打印本页]
作者: 用户云卷云舒    时间: 2024-11-8 00:59
标题: Flink入门实战详解
Flink入门实战

Flink项目构建

1)基于Maven+Idea创建项目：
使用maven进行项目构建，如图1所示。

图-34 构建maven项目

输入项目中的maven的坐标和存储坐标，如图2所示。

图2 maven坐标和存储位置

2)Maven依赖：
  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.12</scala.version>
  6.     <scala.compat.version>2.11</scala.compat.version>
  7.     <hadoop.version>2.6.0</hadoop.version>
  8.     <flink.version>1.9.1</flink.version>
  9. </properties>
  10. <dependencies>
  11.     <dependency>
  12.         <groupId>org.apache.flink</groupId>
  13.         <artifactId>flink-java</artifactId>
  14.         <version>${flink.version}</version>
  15.     </dependency>
  16.     <dependency>
  17.         <groupId>org.apache.flink</groupId>
  18.         <artifactId>flink-streaming-java_2.11</artifactId>
  19.         <version>${flink.version}</version>
  20.     </dependency>
  21.     <dependency>
  22.         <groupId>org.apache.flink</groupId>
  23.         <artifactId>flink-scala_2.11</artifactId>
  24.         <version>${flink.version}</version>
  25.     </dependency>
  26.     <dependency>
  27.         <groupId>org.apache.flink</groupId>
  28.         <artifactId>flink-streaming-scala_2.11</artifactId>
  29.         <version>${flink.version}</version>
  30.     </dependency>
  31.     <dependency>
  32.         <groupId>org.apache.flink</groupId>
  33.         <artifactId>flink-connector-kafka-0.10_2.11</artifactId>
  34.         <version>${flink.version}</version>
  35.     </dependency>
  36.     <dependency>
  37.         <groupId>mysql</groupId>
  38.         <artifactId>mysql-connector-java</artifactId>
  39.         <version>8.0.23</version>
  40.     </dependency>
  41.     <!-- flink-2-hadoop-->
  42.     <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-shaded-hadoop-2 -->
  43.     <dependency>
  44.         <groupId>org.apache.flink</groupId>
  45.         <artifactId>flink-shaded-hadoop-2</artifactId>
  46.         <version>2.7.5-9.0</version>
  47.     </dependency>
  48.     <!-- lombok -->
  49.     <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
  50.     <dependency>
  51.         <groupId>org.projectlombok</groupId>
  52.         <artifactId>lombok</artifactId>
  53.         <version>1.18.12</version>
  54.     </dependency>
  55.     <!-- https://mvnrepository.com/artifact/org.apache.bahir/flink-connector-redis -->
  56.     <dependency>
  57.         <groupId>org.apache.bahir</groupId>
  58.         <artifactId>flink-connector-redis_2.11</artifactId>
  59.         <version>1.0</version>
  60.     </dependency>
  61.     <dependency>
  62.         <groupId>org.apache.flink</groupId>
  63.         <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
  64.         <version>${flink.version}</version>
  65.     </dependency>
  66.     <dependency>
  67.         <groupId>org.apache.flink</groupId>
  68.         <artifactId>flink-table</artifactId>
  69.         <type>pom</type>
  70.         <version>${flink.version}</version>
  71.     </dependency>
  72.     <dependency>
  73.         <groupId>org.apache.flink</groupId>
  74.         <artifactId>flink-table-planner_2.11</artifactId>
  75.         <version>${flink.version}</version>
  76.     </dependency>
  77.     <dependency>
  78.         <groupId>org.apache.flink</groupId>
  79.         <artifactId>flink-table-api-java-bridge_2.11</artifactId>
  80.         <version>${flink.version}</version>
  81.     </dependency>
  82.     <dependency>
  83.         <groupId>org.apache.flink</groupId>
  84.         <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
  85.         <version>${flink.version}</version>
  86.     </dependency>
  87.     <dependency>
  88.         <groupId>org.apache.flink</groupId>
  89.         <artifactId>flink-table-common</artifactId>
  90.         <version>${flink.version}</version>
  91.     </dependency>
  92.     <dependency>
  93.         <groupId>org.apache.flink</groupId>
  94.         <artifactId>flink-cep-scala_2.11</artifactId>
  95.         <version>${flink.version}</version>
  96.     </dependency>
  97.     <dependency>
  98.         <groupId>com.alibaba</groupId>
  99.         <artifactId>fastjson</artifactId>
  100.         <version>1.2.58</version>
  101.     </dependency>
  102.     <dependency>
  103.         <groupId>ru.yandex.clickhouse</groupId>
  104.         <artifactId>clickhouse-jdbc</artifactId>
  105.         <version>0.2.4</version>
  106.         <exclusions>
  107.             <exclusion>
  108.                 <groupId>com.fasterxml.jackson.core</groupId>
  109.                 <artifactId>jackson-databind</artifactId>
  110.             </exclusion>
  111.         </exclusions>
  112.     </dependency>
  113. </dependencies>
  114. <build>
  115.     <sourceDirectory>src/main/scala</sourceDirectory>
  116.     <testSourceDirectory>src/test/scala</testSourceDirectory>
  117.     <plugins>
  118.         <plugin>
  119.             <groupId>org.apache.maven.plugins</groupId>
  120.             <artifactId>maven-compiler-plugin</artifactId>
  121.             <version>2.5.1</version>
  122.             <configuration>
  123.                 <source>${maven.compiler.source}</source>
  124.                 <target>${maven.compiler.target}</target>
  125.                 <!--<encoding>${project.build.sourceEncoding}</encoding>-->
  126.             </configuration>
  127.         </plugin>
  128.         <plugin>
  129.             <groupId>net.alchim31.maven</groupId>
  130.             <artifactId>scala-maven-plugin</artifactId>
  131.             <version>3.2.0</version>
  132.             <executions>
  133.                 <execution>
  134.                     <goals>
  135.                         <goal>compile</goal>
  136.                         <goal>testCompile</goal>
  137.                     </goals>
  138.                     <configuration>
  139.                         <args>
  140.                             <!--<arg>-make:transitive</arg>-->
  141.                             <arg>-dependencyfile</arg>
  142.                             <arg>${project.build.directory}/.scala_dependencies</arg>
  143.                         </args>
  144.                     </configuration>
  145.                 </execution>
  146.             </executions>
  147.         </plugin>
  148.         <plugin>
  149.             <groupId>org.apache.maven.plugins</groupId>
  150.             <artifactId>maven-surefire-plugin</artifactId>
  151.             <version>2.18.1</version>
  152.             <configuration>
  153.                 <useFile>false</useFile>
  154.                 <disableXmlReport>true</disableXmlReport>
  155.                 <includes>
  156.                     <include>**/*Test.*</include>
  157.                     <include>**/*Suite.*</include>
  158.                 </includes>
  159.             </configuration>
  160.         </plugin>
  161.         <plugin>
  162.             <groupId>org.apache.maven.plugins</groupId>
  163.             <artifactId>maven-shade-plugin</artifactId>
  164.             <version>2.3</version>
  165.             <executions>
  166.                 <execution>
  167.                     <phase>package</phase>
  168.                     <goals>
  169.                         <goal>shade</goal>
  170.                     </goals>
  171.                     <configuration>
  172.                         <filters>
  173.                             <filter>
  174.                                 <artifact>*:*</artifact>
  175.                                 <excludes>
  176.                                     <!--
  177.                                     zip -d learn_spark.jar META-INF/*.RSA META-INF/*.DSA META-INF/*.SF
  178.                                     -->
  179.                                     <exclude>META-INF/*.SF</exclude>
  180.                                     <exclude>META-INF/*.DSA</exclude>
  181.                                     <exclude>META-INF/*.RSA</exclude>
  182.                                 </excludes>
  183.                             </filter>
  184.                         </filters>
  185.                         <transformers>
  186.                             <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  187.                                 <mainClass>chapter1.BatchWordCount</mainClass>
  188.                             </transformer>
  189.                         </transformers>
  190.                     </configuration>
  191.                 </execution>
  192.             </executions>
  193.         </plugin>
  194.     </plugins>
  195. </build>
复制代码
Flink基础API概念

Flink编程是在分布式集合的基础的规律的编程模型（好比，执行filtering,mapping,updating,state,joining,grouping,defining,windows,aggregating）。这些集合可以通过外部数据源（好比从文件，kafka的topics、本地大概内存的集合）。通过下沉算子返回结果，好比将数据写入到一个分布式的文件中，大概控制台。Flink程序可以基于各种context、stanalone大概嵌入其他程序进行运行。可以在本地的jvm大概在多台呆板间分布式运行。
基于外部的数据源，好比有界大概无界的数据源，我们大概会选择使用批处理的DataSet API大概流处理的DataStream API来处理。
需要注意的是，在DataStream和DataSet中的绝大多数的API是同等的，只需要替换对应的ExecutionEnvironment大概StreamExecutionEnvironment即可。
Flink在编程的过程中使用特定类——DataSet和DataStream来表现数据，类似Spark中的RDD。可以将其认为是一个可以拥有重复的不可变的集合。其中DataSet表现的是一个有界的数据集，DataStream则表现的是无界的集合。
这些集合在一些关键的地方和Java中的普通集合不同。首先，DataSet和DataStream是不可变的，这就意味着一旦被创建，便不能进行add大概remove的操纵。同样也不能简单的查看集合内部的元素。
Flink可以通过外部的数据源来创建DataSet大概DataStream，也可以通过在一个已知的集合上面执行一系列的Transformation操纵来转换产生新的集合。
Flink程序看起来就是一个普通的程序，进行数据的转换，每一个程序包罗如下相同的集合基础概念，通用编程步调如下：
1)创建一个执行环境ExecutionEnvironment。
2)加载大概创建初始化数据——DataSet大概DataStream。
3)在此数据基础之上进行特定的转化操纵。
4)将计算的结果输出到特定的目的地。
5)触发作业的执行。
Flink编程的入口，便是ExecutionEnvironment，不同之处在于，DataSet和DataStream使用的ExecutionEnvironment不同。DataSet使用ExecutionEnviroment，而DataStream使用StreamExectionEnvironment。
得到ExecutionEnvironment可以通过ExecutionEnvironment的如下方法：
getExecutionEnvironment()
createLocalEnvironment()
createRemoteEnvironment(String host, int port, String... jarFiles)
通常环境下，我们只需要使用getExecutionEnvironment()即可，因为这种方式会自动选择正确的context。假如我们在IDE中执行，则会创建一个Local的Context，假如打包到集群中执行，会返回一个Cluster的Context。
加载数据源的方式有多种。可以一行一个的读入，好比CSV文件，大概自定义格式。假如只是从一个文本文件中按次序读取行数据。只需要如下操纵即可。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> text = env.readTextFile("file:///path")
创建了一个DataStream大概DataSet，接下来便可以执行各种transformation转换操纵。好比执行一个map操纵。
创建一个新的DataStream，类型为Integer的集合。
DataStream中包罗了最终的结果，我们可以将结果通过创建一个sink操纵，写入外部系统中，好比：writeAsText(path)
一旦我们完成整个程序，我们需要通过调用StreamExecutionEnvironment的execute()方法来触发作业的执行。基于ExecutionEnvironment会在本地大概集群中执行。
execute()方法返回值为JobExecutionResult，包罗本次执行时间大概累加器结果信息。
与Spark中的Transformation操纵相同，Flink中的Transformation操纵是Lazy懒加载的，需要execute()去触发。基于此，我们可以创建并添加程序的执行计划。进行任务调度和数据分离，执行更加高效。
目前Flink支持7种数据类型，分别为：
1)Java Tuples和Scala Case Classes。
2)Java POJOS(一种数据结构类型)。
3)Primitive Types(Java的根本数据类型)。
4)Regular Classes(普通类)。
5)Values。
6)Hadoop Writables。
7)SpecialTypes。
DataSet批处理API

Flink中的DataSet程序是实现数据集转换的常规程序（例如，Filter，映射，连接， 分组）。数据集最初是从某些来源创建的（例如，通过读取文件或从本地集合创建）。结果通过接收器返回，接收器可以例如将数据写入（分布式）文件或标准输出（例如命令行终端）。Flink程序可以在各种环境中运行，独立运行或嵌入其他程序中。执行可以在本地JVM中执行，也可以在很多计算机的集群上执行。
  1. import org.apache.flink.api.scala._
  3. object WordCountOps {
  5. def main(args: Array[String]): Unit = {
  7. val env = ExecutionEnvironment.getExecutionEnvironment
  9. val text = env.fromElements("Who's there?",
  11. "I think I hear them. Stand, ho! Who's there?"
  13. )
  15. val wordCounts:DataSet[(String, Int)] = text
  17. .flatMap(line => line.split("\\s+")).map((_, 1))
  19. .groupBy(0)
  21. .sum(1)
  23. wordCounts.print()
  25. }
  27. }
复制代码
Streaming流式处理API

Flink中的DataStream程序是实现数据流转换的常规程序（例如 filtering, updating state, defining windows, aggregating）。最初从各种源（例如， message queues, socket streams, files）创建数据流。结果通过接收器返回，接收器可以例如将数据写入文件或标准输出（例如命令行终端）。Flink程序可以在各种环境中运行，独立运行或嵌入其他程序中。执行可以在本地JVM中执行，也可以在很多计算机的集群上执行。
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}
object StreamDemo {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val file = env.socketTextStream("localhost", 9999)
        val spliFile: DataStream[String] = file.flatMap(_.split(" "))
        val wordAndOne: DataStream[(String, Int)] = spliFile.map((_, 1))
        val keyed = wordAndOne.keyBy(data=>data._1)
        val wordAndCount: DataStream[(String, Int)] = keyed.sum(1)
        wordAndCount.print()
        env.execute()
    }
}
要运行示例程序，首先从终端使用netcat启动输入流：
nc -lk 9999
只需键入一些单词就可以返回一个新单词。这些将是字数统计程序的输入。
Flink程序提交到集群

1)Web提交方式：

图3 web提交方式

1)脚本方式：
  1. #!/bin/sh
  3. FLINK_HOME=/home/bigdata/apps/flink
  5. $FLINK_HOME/bin/flink run \
  7. -c BatchDemo \
  9. /root/wc.jar \
  11. hdfs://hadoop101:8020/wordcount/words.txt \
  13. hdfs://hadoop101:8020/wordcount/output3
复制代码


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4