【Flink】Flink流应用开发

痛定思痛。 2023-07-13 04:24 33阅读 0赞

## 1、Flink概述 ##

（1）Flink简介  
Flink是一个低延迟、高吞吐、统一的大数据分布式实时计算引擎，使用官网的一句话来介绍Flink就是“Stateful Computations Over Streams”（有状态的流式计算）。  
（2）Flink相对于Spark Streaming的特点  
1）Checkpoint机制，即容错机制  
2）State状态化机制，即状态管理机制  
3）Time时间机制  
4）Window语义  
5）时效性短  
6）架构不同  
（3）Flink的应用场景  
1）实时数据仓库：数据的实时清洗、归并、结构化，数据仓库的补充和优化。  
2）实时监控：对用户行为或相关事件进行实时监测和分析，基于风控规则进行预警，用户行为预警、app crash预警、服务器攻击预警。  
3）流数据分析：实时计算相关指标反馈及时调整决策，内容投放，无线智能推送，实时个性化推荐。  
4）实时报表：双11、双12等活动直播大屏，对外数据产品：生意参谋等，数据化运营。  
（4）Flink系统架构  
Flink是Master-Slave架构，Master是JobManager，Slave是TaskManager。

## 2、Flink DataStream编程 ##

（1）程序结构

object WordCount{
    	def main(args: Array[String]): Unit = {
    		// 1.设置运行环境
    		val env = StreamExecutionEnvironment.getExecutionEnvironment
    		// 2.配置数据源，读取数据
    		val ds = env.socketTextStream("127.0.0.1", 7777)
    		// 3.进行一系列转换操作
    		ds.flatMap(line => line.split("\\s+")
    		  .map(word => (word, 1)))
    		  .keyBy(0)
    		  .timeWindow(Time.seconds(10))
    		  .sum(1)
    		// 4.配置数据外部存储，并写出数据
    		ds.print()
    		// 5.提交执行
    		env.execute(getClass.getSimpleName)
    	}
    }

产生数据：

object OrderDataProducer {
    	val productNames = Array("iphone8p", "mbp2018", "locklock cup", "dell computer", "iphone7p")
    	val amounts = Array(6888, 16888, 25.3, 6500, 7000)
    	val sensor = Array("a", "b", "c", "d", "e")
    	val random = new Random()
    
    	def main(args: Array[String]): Unit = {
    		testSensorData
    		// testOrderData2
    		// testProductData()
    	}
    
    	def testProductData(): Unit = {
    		val i = 1
    		// {"productId": "p1001", "productName": "Apple", "amount": 8000, "createTime": "2019-08-20 12:00:01"}
    		while(i < 1000)
    		val productObj = new JSONObject()
    		productObj.put("productId", "p300"+i)
    		productObj.put("productName", "Apple"+i)
    		productObj.put("amount", random.nextInt(3000))
    		productObj.put("createTime", DateUtils.getNowDatetimeNoms())
    		println(productObj.toString())
    		Thread.sleep(1000)
    		KafkaUtils.produceData("flink-test-products", productObj.toString())
    		i += 1
    	}
    }
    
    def testSensorData(): Unit = {
    	var i = 1
    	while(i < 100){
    		val sb = new mutable.StringBuilder()
    		// sb.append(sensor(random.nextInt(4))).append(".")
    		sb.append("a").append(",")
    		sb.append("i").append(",")
    		sb.append(1 + random.nextInt(100)).append(",")
    		sb.append(DateUtils.getNowDatetime())
    		KafkaUtils.produceData("flink-kafka-test".sb.toString())
    		println(sb.toString())
    		Thread.sleep(200)
    		i += 1
    	}
    }

消费数据：

object RealtimeDemo {
    	def main(args: Array[String]): Unit = {
    		val topicName = "flink-kafka-test"
    		val env = StreamExecutionEnvironment.getExecutionEnvironment
    
    		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    		
    		val props = new Properties()
    		props.setProperty("bootstrap.servers", "localhost:9092")
    		props.put("auto.offset.reset", "latest")
    		props.setProperty("group.id", "flink-kafka-test")
    		props.put("enable.auto.commit", "false")
    
    		env.setParallelism(2)
    		val consumer = new FlinkKafkaConsumer011[String](topicName, new SimpleStringSchema(), props)
    		env.addSource(consumer)
    		   .map(line => line + ">>>>>" + DateUtils.getNowDatetime())
    		   .print()
    		env.execute(getClass.getSimpleName)
    	}
    }

（2）流应用开发步骤  
1）创建运行环境  
2）配置，如：时间、Watermark、并行度、Checkpoint等  
3）创建Source，如Kafka，Redis等  
4）对DataStream进行Transformation操作  
5）输出到Sink，如Redis、HBase等  
6）执行程序  
（3）Flink数据类型  
1）基础类型（Basic）：所有Java的基础类型（包括装箱与未装箱的类型），如：int，double，float，boolean，long，void，Integer，Double，String，Date，BigDecimal，BigInteger等。  
2）数组（Arrays）：基础类型构成的数组或者Object\[\]。  
3）复合类型（Composite）：Flink Java Tuple，Scala Tuple，Row，POJO（必须是public以及不含非静态的内部类，必须有一个public类型的无参构造方法，所有的非静态、非transient字段必须是public的，或者提供public的getter和setter方法）。  
4）辅助类型（Auxiliary）：Option、Either、Lists、Maps等。  
5）泛型和其他类型（Generic）：由Kyro提供序列化支持。

## 3、Flink流应用开发具体步骤 ##

（1）创建运行环境  
有三种API：

1. getExecutionEnvironment()
    
    2. createLocalExecutionEnvironment()
    
    3. createRemoteExecutionEnvironment()

（2）Data Source

<table> 
 <thead> 
  <tr> 
   <th>分类</th> 
   <th>类型</th> 
   <th>描述</th> 
   <th>API</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>内置数据源</td> 
   <td>基于socket</td> 
   <td>监听主机的host port，从Socket中获取数据</td> 
   <td>socketTextStructure()</td> 
  </tr> 
  <tr> 
   <td>内置数据源</td> 
   <td>基于File</td> 
   <td>监听文件修改并读取其内容</td> 
   <td>readTextFile()</td> 
  </tr> 
  <tr> 
   <td>内置数据源</td> 
   <td>基于File</td> 
   <td>读取流式文件</td> 
   <td>readFileStream()</td> 
  </tr> 
  <tr> 
   <td>内置数据源</td> 
   <td>基于集合</td> 
   <td>有界数据集，更偏向于本地测试用</td> 
   <td>fromCollection()/fromElements()</td> 
  </tr> 
  <tr> 
   <td>Connector</td> 
   <td>外部扩展数据源</td> 
   <td>丰富的外部扩展数据源</td> 
   <td>env.addSource(new FlinkKafkaConsumer[String](topicName, new SimpleStringSchema(), prop))</td> 
  </tr> 
  <tr> 
   <td>自定义Source</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>env.addSource(sourceFunction)</td> 
  </tr> 
 </tbody> 
</table>

（3）Connector  
Connectors是数据进出Flink的一套接口和实现，可以实现Flink与各种存储、系统的连接。

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>Source</th> 
   <th>Sink</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Apache Kafka</td> 
   <td>√</td> 
   <td>√</td> 
  </tr> 
  <tr> 
   <td>Apache Cassandra</td> 
   <td>×</td> 
   <td>√</td> 
  </tr> 
  <tr> 
   <td>Elasticsearch</td> 
   <td>×</td> 
   <td>√</td> 
  </tr> 
  <tr> 
   <td>Hadoop File System</td> 
   <td>×</td> 
   <td>√</td> 
  </tr> 
  <tr> 
   <td>RabbitMQ</td> 
   <td>√</td> 
   <td>√</td> 
  </tr> 
  <tr> 
   <td>Apache ActiveMQ</td> 
   <td>√</td> 
   <td>√</td> 
  </tr> 
  <tr> 
   <td>Apache Flume</td> 
   <td>√</td> 
   <td>√</td> 
  </tr> 
  <tr> 
   <td>Redis</td> 
   <td>×</td> 
   <td>√</td> 
  </tr> 
 </tbody> 
</table>

（4）自定义Source  
自定义数据实现方式：  
1）继承SourceFunction抽象类（非并行数据源）  
2）继承ParallelSourceFunction抽象类（并行数据源）  
3）继承RichParallelSourceFunction抽象类（并行数据源，额外提供open和close方法）  
（5）Transformations操作  
1）基于单条记录的map，filter等操作  
2）基于窗口的window操作  
3）合并多条流为单条流的union，join，connect操作  
4）拆分单条流为多条流的split操作  
（6）窗口模型  
窗口模型就是将流做一定范围内的统计。  
Window分为Time Window，Count Window，自定义Window。  
Time Window又包含Tumbling Windows（翻转窗口，如1点、2点、3点…分别统计一次）和Sliding Windows（滑动窗口，如每隔5秒统计一次）。  
Count Window也包含Tumbling Windows（翻转窗口，如1000个、2000个、3000个统计一次）和Sliding Windows（滑动窗口，如每5个统计一次）。  
（7）常用算子

<table> 
 <thead> 
  <tr> 
   <th>算子名称</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>map</td> 
   <td>可以理解为映射，对每个元素进行一定的变换后，映射为另一个元素</td> 
  </tr> 
  <tr> 
   <td>flatmap</td> 
   <td>可以理解为将元素摊平，每个元素可以变为0个、1个、或者多个元素</td> 
  </tr> 
  <tr> 
   <td>filter</td> 
   <td>筛选过滤</td> 
  </tr> 
  <tr> 
   <td>keyBy</td> 
   <td>逻辑上将Stream根据指定的Key进行分区，是根据key的散列值进行分区的</td> 
  </tr> 
  <tr> 
   <td>reduce</td> 
   <td>归并操作，它可以将KeyedStream转变为DataStream</td> 
  </tr> 
  <tr> 
   <td>fold</td> 
   <td>给定一个初始值，将各个元素逐个归并计算，它将KeyedStream转变为DataStream</td> 
  </tr> 
  <tr> 
   <td>union</td> 
   <td>可以将多个流合并到一个流中，以便对合并的流进行统一处理，是对多个流的水平拼接</td> 
  </tr> 
  <tr> 
   <td>join</td> 
   <td>根据指定的key将两个流进行关联</td> 
  </tr> 
  <tr> 
   <td>coGroup</td> 
   <td>关联两个流，关联不上的也保留下来</td> 
  </tr> 
  <tr> 
   <td>split</td> 
   <td>将一个流拆分为多个流</td> 
  </tr> 
  <tr> 
   <td>window</td> 
   <td>按时间进行聚合或者其他条件对KeyedStream进行分组</td> 
  </tr> 
 </tbody> 
</table>

（8）Data Sink

<table> 
 <thead> 
  <tr> 
   <th>分类</th> 
   <th>类型</th> 
   <th>描述</th> 
   <th>API</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>内置Sink</td> 
   <td>文件</td> 
   <td>以文件的形式输出</td> 
   <td>writeAsText()</td> 
  </tr> 
  <tr> 
   <td>内置Sink</td> 
   <td>打印</td> 
   <td>监听文件修改并读取其内容</td> 
   <td>print()/printErr()</td> 
  </tr> 
  <tr> 
   <td>内置Sink</td> 
   <td>流</td> 
   <td>以流的形式输出，要求定义serializationSchema</td> 
   <td>writeToSocket()</td> 
  </tr> 
  <tr> 
   <td>Connector</td> 
   <td>外部扩展数据源</td> 
   <td>丰富的外部扩展数据源</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>自定义Sink</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>DataStream.addSink(sinkFunction)</td> 
  </tr> 
 </tbody> 
</table>

（9）运行  
两种运行方式：  
1）直接在IDEA中运行  
2）提交到集群

nohup bin/flink run -c com.gupao.bd.sample.flink.realtime.app.ProductDataStreamSync /cloud/code/gp/gp-bd/demo_flink/target/demo_flink-1.1-SNAPSHOT-jar-with-dependencies.jar &