【Flink-scala】DataSet编程模子先容及数据源

DataStream 学习

1.DataStream编程模子总结

  
<hr> 先容

Flink把批处理处罚看成是一个流处理处罚的特例，因此可以在底层同一的流处理处罚引擎上，同时提供了STREAM API和SET API,经典的有限数据流处理处罚方式有：

由于批处理处罚的对象是有界数据集，因此批处理处罚不须要时间和窗口机制
一、DataSet编程模子

link批处理处罚步伐的根本运行流程包罗以下4个步调：

  

• 创建实行情况；  
  
• 创建数据源；  
  
• 指定对数据举行的转换操纵；  
  
• 指定命据盘算的输出效果方式。
  上面第1步中创建批处理处罚实行情况的方式如下：
  

1. val env = ExecutionEnvironment.getExecutionEnvironment

复制代码

别的，还须要在pom.xml文件中引入flink-scala_2.12依靠库，具体如下：

1. <dependency>
2.             <groupId>org.apache.flink</groupId>
3.             <artifactId>flink-scala_2.12</artifactId>
4.             <version>1.11.2</version>
5. </dependency>

复制代码

编程模子如图：

数据的处理处罚过程：

读取数据源-举行转换操纵-获取效果数据。
批处理处罚数据的根本流程：

二、数据源

1.文件类数据源

Flink提供了从文件中读取数据天生DataSet的多种方法，具体如下：
readTextFile(path):逐行读取文件并将文件内容转换成DataSet范例数据集；
readTextFileWithValue(path)：读取文本文件内容，并将文件内容转换成DataSet[StringValue]范例数据集。
该方法与readTextFile(String)差别的是，其泛型是StringValue，是一种可变的String范例，通过StringValue存储文本数据可以有用低沉String对象创建数目，减小垃圾接纳的压力；
readCsvFile(path)：分析以逗号(或其他字符)分隔字段的文件，返回元组或POJO对象；
readSequenceFile(Key, Value, path)：读取SequenceFile，以Tuple2<Key, Value>范例返回。
以readTextFile(path)为例，可以利用如下语句读取文本文件内容：

1. val dataSet : DataSet[String] = env.readTextFile("file:///home/hadoop/word.txt")

复制代码

假设有一个CSV格式文件sales.csv，内容如下：

1. transactionId,customerId,itemId,amountPaid
2. 111,1,1,100.0
3. 112,2,2,505.0
4. 113,1,3,510.0
5. 114,2,4,600.0
6. 115,3,2,500.0

复制代码

则可以利用如下步伐读取该CSV文件：

1. import org.apache.flink.api.scala.ExecutionEnvironment
2. import org.apache.flink.api.scala._
4. object ReadCSVFile{
5.    
6.    
7.   def main(args: Array[String]): Unit = {
8.    
9.    
10.     val bEnv = ExecutionEnvironment.getExecutionEnvironment
11.     val filePath="file:///home/hadoop/sales.csv"
12.     val csv = bEnv.readCsvFile[SalesLog](filePath,ignoreFirstLine = true)//这里
13.     csv.print()
14.   }
15.   case class SalesLog(transactionId:String,customerId:String,itemId:String,amountPaid:Double)//这里定义的类型
16. }

复制代码

效果如下：

1. SalesLog(111,1,1,100.0)
2. SalesLog(112,2,2,505.0)
3. SalesLog(113,1,3,510.0)
4. SalesLog(114,2,4,600.0)
5. SalesLog(115,3,2,500.0)

复制代码

2.聚集类数据源

Flink提供了fromCollection()、fromElements()和generateSequence()等方法，来构建聚集类数据源，具体如下：
fromCollection()：从聚集中创建DataSet数据集，聚集中的元素数据范例类似；
fromElements()：从给定命据元素序列中创建DataSet数据集，且全部的数据对象范例必须划一；
generateSequence()：指定一个范围区间，然后在区间内部天生数字序列数据集,由于是并行处理处罚的，以是终极的次序不能包管划一。

1. val myArray = Array("hello world","hadoop spark flink")
2. val collectionSet = env.fromCollection(myArray)//从集合中获取
4. val dataSet = env.fromElements("hadoop","spark","flink")//一个个元素获取
6. val numSet = env.generateSequence(1,10)//生成的数据 1 2 3 4 ... 10   包含10

复制代码

3.通用类数据