Flink-DataStream执行环境和数据读取

编辑执行环境

创建执行环境

执行模式

触发程序执行

源算子（Source）

读取有界数据流

读取无界数据

读取自定义数据源（源算子）

DataStream是一个 Flink 程序，其实就是对 DataStream 的各种转换。具体来说，代码基本上都由以下几部分构成：

获取执行环境（Execution Environment）
读取数据源（Source）
定义基于数据的转换操作（Transformations）
定义计算结果的输出位置（Sink）
触发程序执行（Execute）

执行环境

创建执行环境

编写Flink 程序的第一步，就是创建执行环境。我们要获取的执行环境，是 StreamExecutionEnvironment 类的对象，这是所有 Flink 程序的基础。在代码中创建执行环境的方式，就是调用这个类的静态方法，具体有以下三种:

1、getExecutionEnvironment（智能）

就是直接调用 getExecutionEnvironment 方法。它会根据当前运行的上下文直接得到正确的结果；也就是说，这个方法会根据当前运行的方式，自行决定该返回什么样的运行环境。

val env = StreamExecutionEnvironment.getExecutionEnvironment

这种“智能”的方式不需要我们额外做判断，用起来简单高效，是最常用的一种创建执行环境的方式。

2、createLocalEnvironment（本地）

这个方法返回一个本地执行环境。可以在调用时传入一个参数，指定默认的并行度；如果不传入，则默认并行度就是本地的 CPU 核心数。

val localEnvironment = StreamExecutionEnvironment.createLocalEnvironment()

3.、createRemoteEnvironment（远程）

这个方法返回集群执行环境。需要在调用时指定 JobManager 的主机名和端口号，并指定要在集群中运行的 Jar 包。

val remoteEnv = StreamExecutionEnvironment.createRemoteEnvironment("host", // JobManager 主机名1234, // JobManager 进程端口号"path/to/jarFile.jar" // 提交给 JobManager 的 JAR 包
)

执行模式

在之前的 Flink 版本中，批处理的执行环境与流处理类似，是调用类 ExecutionEnvironment 的静态方法，并返回它的对象：

// 批处理环境
val batchEnv = ExecutionEnvironment.getExecutionEnvironment
// 流处理环境
val env = StreamExecutionEnvironment.getExecutionEnvironment

流执行模式（STREAMING）这是 DataStream API 最经典的模式，一般用于需要持续实时处理的无界数据流。默认情况下，程序使用的就是 STREAMING 执行模式。
批执行模式（BATCH）专门用于批处理的执行模式, 这种模式下，Flink 处理作业的方式类似于 MapReduce 框架。对于不会持续计算的有界数据，我们用这种模式处理会更方便。
自动模式（AUTOMATIC）在这种模式下，将由程序根据输入数据源是否有界，来自动选择执行模式。由于 Flink 程序默认是 STREAMING 模式，我们这里重点介绍一下 BATCH 模式的配置。

BATCH （批处理）模式的配置，主要有两种方式：

1、通过命令行配置

在提交作业时，增加 execution.runtime-mode 参数，指定值为 BATCH。

bin/flink run -Dexecution.runtime-mode=BATCH ...

2、通过代码配置（不推荐）

在代码中，直接基于执行环境调用 setRuntimeMode 方法，传入 BATCH 模式。

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setRuntimeMode(RuntimeExecutionMode.BATCH)

触发程序执行

我们需要显式地调用执行环境的 execute()方法，来触发程序执行。execute()方法将一直等待作业完成，然后返回一个执行结果（JobExecutionResult）。

env.execute()

源算子（Source）

Flink 可以从各种来源获取数据，然后构建 DataStream 进行转换处理。一般将数据的输入来源称为数据源，而读取数据的算子就是源算子（Source）。所以，Source 就是我们整个处理程序的输入端。

Flink 代码中通用的添加 Source 的方式，是调用执行环境的 addSource()方法，方法传入一个对象参数，需要实现 SourceFunction 接口，返回一个 DataStream。

//通过调用 addSource()方法可以获取 DataStream 对象
val stream = env.addSource(...)

读取有界数据流

案例需求：

为了更好地理解，我们先构建一个实际应用场景。比如网站的访问操作，可以抽象成一个三元组（用户名，用户访问的 url，用户访问 url 的时间戳），所以在这里，我们可以创建一个类 Event，将用户行为包装成它的一个对象。Event 包含了以下一些字段

定义样例类

case class Event(user: String, url: String, timestamp: Long)

1、从元素中读取数据

//从元素中读取数据val stream: DataStream[Int] = env.fromElements(1, 2, 3, 4, 5, 6) //指定元素数据//创建当前样例类Event的对象val value1: DataStream[Event] = env.fromElements(Event("张三", "01", 1000L),Event("李四", "04", 2000L),Event("王五", "01", 6000L),Event("赵六", "03", 1000L))

2、从集合中读取数据

 //从集合中读取数据//定义一个集合val e = List(Event("张三", "01", 1000L),Event("李四", "04", 2000L),Event("王五", "01", 6000L),Event("赵六", "03", 1000L))val value2: DataStream[Event] = env.fromCollection(e)

3、从文件读取数据

真正的实际应用中，自然不会直接将数据写在代码中。通常情况下，我们会从存储介质中获取数据，一个比较常见的方式就是读取日志文件。这也是批处理中最常见的读取方式。

val value3: DataStream[String] = env.readTextFile("datas\\wc.txt")

读取无界数据

1、从 Socket 读取数据

不论从集合还是文件，我们读取的其实都是有界数据。在流处理的场景中，数据往往是无界的。一个简单的例子，就是我们之前用到的读取 socket 文本流。这种方式由于吞吐量小、稳定性较差，一般也是用于测试。

val stream = env.socketTextStream("localhost", 7777)

2、从 Kafka 读取数据

Flink 官方提供了连接工具 flink-connector-kafka，直接帮我们实现了一个消费者 FlinkKafkaConsumer，它就是用来读取 Kafka 数据的 SourceFunction。

添加连接器依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka_${scala.binary.version}</artifactId><version>${flink.version}</version></dependency>

调用 env.addSource()，传入 FlinkKafkaConsumer 的对象实例即可：

def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//设置全局并行度env.setParallelism(1)//使用Properties对象保存kafa连接的相关配置val properties: Properties = new Properties()properties.setProperty("bootstrap.servers","master:9000")properties.setProperty("group.id","consumer-group")val stream: DataStream[String] = env.addSource( new FlinkKafkaConsumer[String]("clicks", new SimpleStringSchema(),properties))stream.print()env.execute()}

读取自定义数据源（源算子）

接下来我们创建一个自定义的数据源，实现 SourceFunction 接口。主要重写两个关键方法： run()和 cancel()。

run()方法：使用运行时上下文对象（SourceContext）向下游发送数据；
cancel()方法：通过标识位控制退出循环，来达到中断数据源的效果。

创建自定义读取数据源类

class f4 extends SourceFunction[ Event ]{ //实现SourceFunction接口 泛型为之前定义好的样例类Event//标志位var running = true//重写抽象方法override def run(sourceContext: SourceFunction.SourceContext[Event]): Unit = {//随机数生成器val random = new Random ()//定义数据随机选择的范围val user = Array ("张三", "李四", "王五")val url = Array ("02", "01", "03", "04")//用标志位作为循环判断条件，不停的发出数据while (running) {val event = Event (user (random.nextInt (user.length) ), url (random.nextInt (url.length) ), Calendar.getInstance.getTimeInMillis)//调用ctx的方法向下游发送数据sourceContext.collect (event)//每隔1秒发送一条数据Thread.sleep (1000)}}override def cancel(): Unit = running = false
}

测试类运行

def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//全局并行度设置为1env.setParallelism(1)//读取自定义的数据源val ste: DataStream[Event] = env.addSource(new f4)//输出ste.print()//执行env.execute()}

执行效果：每隔1秒自动输出一行数据