入门案例

创建项目

1. 设置MAVEN Repository仓库目录及Maven安装目录

   

2. 创建整个Flink基础课程Maven Project，

   

3. 添加scala支持

   

   

4. 在src/main下面创建scala文件夹，并标注为Sources Root

5. 在项目根目录下创建datas文件夹，并将words拷贝到当前目录下。

6. 配置pom.xml文件

   ​x
   <?xml version="1.0" encoding="UTF-8"?>
   <project xmlns="http://maven.apache.org/POM/4.0.0"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
       <modelVersion>4.0.0</modelVersion>
       <groupId>com.itbaizhan</groupId>
       <artifactId>flink-demo</artifactId>
       <version>1.0.0</version>
   
       <properties>
           <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
           <maven.compiler.source>1.8</maven.compiler.source>
           <maven.compiler.target>1.8</maven.compiler.target>
       </properties>
       <dependencies>
           <dependency>
               <groupId>org.apache.flink</groupId>
               <artifactId>flink-clients_2.12</artifactId>
               <version>1.14.6</version>
           </dependency>
           <dependency>
               <groupId>org.apache.flink</groupId>
               <artifactId>flink-scala_2.12</artifactId>
               <version>1.14.6</version>
           </dependency>
           <dependency>
               <groupId>org.apache.flink</groupId>
               <artifactId>flink-streaming-scala_2.12</artifactId>
               <version>1.14.6</version>
           </dependency>
           <dependency>
               <groupId>org.apache.flink</groupId>
               <artifactId>flink-runtime-web_2.12</artifactId>
               <version>1.14.6</version>
           </dependency>
           <dependency>
               <groupId>org.slf4j</groupId>
               <artifactId>slf4j-api</artifactId>
               <version>1.7.7</version>
               <scope>runtime</scope>
           </dependency>
           <dependency>
               <groupId>org.slf4j</groupId>
               <artifactId>slf4j-log4j12</artifactId>
               <version>1.7.7</version>
               <scope>runtime</scope>
           </dependency>
           <dependency>
               <groupId>log4j</groupId>
               <artifactId>log4j</artifactId>
               <version>1.2.17</version>
               <scope>runtime</scope>
           </dependency>
       </dependencies>
       <build>
           <plugins>
               <!-- 在maven项目中既有java又有scala代码时配置 maven-scala-plugin 插件打包时可以将两类代码一起打包 -->
               <plugin>
                   <groupId>org.scala-tools</groupId>
                   <artifactId>maven-scala-plugin</artifactId>
                   <version>2.15.2</version>
                   <executions>
                       <execution>
                           <goals>
                               <goal>compile</goal>
                               <goal>testCompile</goal>
                           </goals>
                       </execution>
                   </executions>
               </plugin>
               <!-- maven 打jar包需要插件 -->
               <plugin>
                   <groupId>org.apache.maven.plugins</groupId>
                   <artifactId>maven-assembly-plugin</artifactId>
                   <version>3.1.0</version>
                   <configuration>
                       <!--<appendAssemblyId>false</appendAssemblyId>-->
                       <descriptorRefs>
                           <descriptorRef>jar-with-dependencies</descriptorRef>
                       </descriptorRefs>
                       <archive>
                           <manifest>
                               <mainClass>com.itbaizhan.scala.flink.xxx</mainClass>
                           </manifest>
                       </archive>
                   </configuration>
                   <executions>
                       <execution>
                           <id>make-assembly</id>
                           <phase>package</phase>
                           <goals>
                               <goal>single</goal>
                           </goals>
                       </execution>
                   </executions>
               </plugin>
           </plugins>
       </build>
   </project>
   

7. 配置log4j.properties文件

   xxxxxxxxxx
   # This affects logging for both user code and Flink
   log4j.rootLogger=ERROR, console
   # log level INFO. The root logger does not override this. You have to manually
   # change the log levels here.
   log4j.logger.akka=ERROR
   log4j.logger.org.apache.kafka=ERROR
   log4j.logger.org.apache.hadoop=ERROR
   log4j.logger.org.apache.zookeeper=ERROR
   
   # Log all infos to the console
   log4j.appender.console=org.apache.log4j.ConsoleAppender
   log4j.appender.console.layout=org.apache.log4j.PatternLayout
   log4j.appender.console.layout.ConversionPattern=%d{MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
   
   # Suppress the irrelevant (wrong) warnings from the Netty channel handler
   log4j.logger.org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, console
   

8. 最终项目结构

   

编程模型

1. Flink API，提供四个层次API，越在下面API，越复杂和灵活；越在上面API，使用越简单和抽象。

   

   

   https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/dev/dataset/overview/

2. 编程模型，无论编写批处理还是流计算程序，分为三个部分：Source、Transformation和Sink

   1、从数据源DataSource获取数据

   流计算：DataStream 重点 ​ 批处理：DataSet 了解

   2、对数据进行转换处理

   3、结果数据输出DataSink

3. 无论批处理Batch，还是流计算Stream，首先需要创建执行环境StreamExecutionEnvironment对象，类似Spark中SparkSession。

   基于Flink计算引擎，分别实现批处理（Batch）和流计算（Streaming ）中：词频统计WordCount。

实时效果反馈

1. 以下不属于Flink编程模型计算程序组成部分是？

A Source

B Transformation

C Action

D Sink

答案：

1=>C