Flink万物之始API流环境获取与Source定义

梦里梦外; 2023-03-13 05:49 2阅读 0赞

经过上几篇的教程我相信你对Flink的认识已经很不错了，我们是时候去开始我们最喜欢的编程环节了！！那么我们在写一个Flink程序的第一步是干什么呢？什么？定义source。很正确，但是少了一点，在定义source前我们需要先获取Flink的执行环境，究竟是本地执行还是集群执行，然后再去定义source再去Transform与sink！！所以我们先开始我们的第一步。

# 一、流处理环境获取 #

## 1.1 getExecutionEnvironment ##

首先肯定要先给大家讲讲常用的创建一个执行环境的方法，getExecutionEnvironment表示当前执行程序的上下文。 如果程序是独立调用的，则此方法返回本地执行环境；如果从命令行客户端调用程序以提交到集群，则此方法返回此集群的执行环境，也就是说，getExecutionEnvironment会根据查询运行的方式决定返回什么样的运行环境，是最常用的一种创建执行环境的方式。

val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    val env = StreamExecutionEnvironment.getExecutionEnvironment

如果没有设置并行度，会以flink-conf.yaml中的配置为准，默认是1。

## 1.2 createLocalEnvironment ##

返回本地执行环境，需要在调用时指定默认的并行度。

val env = StreamExecutionEnvironment.createLocalEnvironment(1)

## 1.3 createRemoteEnvironment ##

返回集群执行环境，将Jar提交到远程服务器。需要在调用时指定JobManager的IP和端口号，并指定要在集群中运行的Jar包。

val env = ExecutionEnvironment.createRemoteEnvironment("jobmanage-hostname", 6123,"YOURPATH//wordcount.jar")

# 二、Source定义 #

讲完了Flink流环境的获取，那么接下来也就是Source的定义了，这里会跟大家讲述两种，一种是kafkaSource，另一种是自定义Source

## 2.1 KafkaSource ##

首先需要在pom文件中引入kafka连接器的依赖：

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
        <version>1.10.0</version>
    </dependency>

具体代码如下：

val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")
    
    val stream3 = env.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))

## 2.2 自定义Source ##

除了Flink已经封装好的source数据来源，其实我们还可以自定义source。需要做的，只是实现一个SourceFunction就可以。具体实现代码如下：

val stream = env.addSource( new MyCustomSource() )
    我们希望可以随机生成传感器数据，MyCustomSource具体的代码实现如下：
    class MyCustomSource extends SourceFunction[BookInfo]{
    
    // flag: 表示数据源是否还在正常运行
    var running: Boolean = true
    
    override def cancel(): Unit = {
    running = false
    }
    
    override def run(ctx: SourceFunction.SourceContext[BookInfo]): Unit = {
    // 初始化一个随机数发生器
    val rand = new Random()
    
    var curTemp = 1.to(10).map(
    i => ( "book" + i, 15 + rand.nextGaussian() * 20 )
    )
    
    while(running){
    // 更新书的价格
    curTemp = curTemp.map(
    t => (t._1, t._2 + rand.nextGaussian() )
    )
    // 获取当前时间戳
    val curTime = System.currentTimeMillis()
    
    curTemp.foreach(
    t => ctx.collect(BookInfo(t._1, curTime, t._2))
    )
    Thread.sleep(100)
    }
    }
    }