Flink8：Flink流处理Api之Source
流处理基本步骤：
Source
1. 从集合读取数据
package wordcount
import org.apache.flink.streaming.api.scala._
// 定义样例类，传感器id，时间戳，温度
case class SensorReading(id:String,timestamp:Long,temperature:Double)
object Sensor {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)
// 1. 从集合中读取数据
val stream1 = env.fromCollection(List(
SensorReading("sensor_1", 1547718199, 35.80018327300259),
SensorReading("sensor_6", 1547718201, 15.402984393403084),
SensorReading("sensor_7", 1547718202, 6.720945201171228),
SensorReading("sensor_10", 1547718205, 38.101067604893444)
))
stream1.print("stream1:").setParallelism(1)
env.execute()
}
}
运行代码，打印结果：
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
stream1:> SensorReading(sensor_1,1547718199,35.80018327300259)
stream1:> SensorReading(sensor_6,1547718201,15.402984393403084)
stream1:> SensorReading(sensor_7,1547718202,6.720945201171228)
stream1:> SensorReading(sensor_10,1547718205,38.101067604893444)
Process finished with exit code 0
2. 从文件读取数据
val stream2 = env.readTextFile("YOUR_FILE_PATH")
3. 以kafka消息队列的数据作为来源
- 首先，pom配置如下：
<?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>org.example</groupId>
<artifactId>flink-study</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_2.11</artifactId>
<version>1.7.2</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId>
<version>1.7.2</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka-0.11_2.11</artifactId>
<version>1.7.2</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.4.6</version>
<executions>
<execution>
<goals>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<!-- <version>3.0.0</version>-->
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
- 集成开发代码：
package com.stream;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import java.util.Properties;
public class KafkaStream {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
Properties prop = new Properties();
prop.setProperty("bootstrap.servers", "localhost:9092")
prop.setProperty("group.id", "consumer-group")
prop.setProperty("key.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer")
prop.setProperty("value.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer")
prop.setProperty("auto.offset.reset", "latest")
DataStreamSource dstream = environment.addSource(new FlinkKafkaConsumer011("senser", new SimpleStringSchema(), prop));
dstream.print("kafka test").setParallelism(1);
environment.execute();
}
}
-
运行，上面开发是重点，因为演示环境麻烦，这里只给出操作步骤：
3.1 在虚拟机启动zookeeper和kafka服务
3.2 在虚拟机启动代码里响应配置的topic的producer
3.3 进入flink主目录下执行./bin/start-cluster.sh启动flink
3.4 assembly方式打包上传jar包到虚拟机执行:
./flink run -c com.stream.KafkaStream flink-study-1.0-SNAPSHOT-jar-with-dependencies.jar
3.5 在producer输入数据，控制台没有结果，可以访问<u>h</u><u>ttp://hadoop1:8081</u>
在TaskManager上看输出日志。 -
Flink+kafka是如何实现exactly-once语义的：
Flink通过checkpoint来保存数据是否处理完成的状态：
由JobManager协调各个TaskManager进行checkpoint存储，checkpoint保存在 StateBackend中，默认StateBackend是内存级的，也可以改为文件级的进行持久化保存。
执行过程实际上是一个两段式提交，每个算子执行完成，会进行“预提交”，直到执行完sink操作，会发起“确认提交”，如果执行失败，预提交会放弃掉。
如果宕机需要通过StateBackend进行恢复，只能恢复所有确认提交的操作。
4. 自定义Source
除了以上的source数据来源，我们还可以自定义source。需要做的，只是传入一个SourceFunction就可以。具体调用如下：
package wordcount
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
object SourceTest {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)
//自定义source
val stream = env.addSource(new MySensorSource())
stream.print("stream").setParallelism(1)
env.execute("source test")
}
}
我们希望可以随机生成传感器数据，MySensorSource具体的代码实现如下：
package wordcount
import org.apache.flink.streaming.api.functions.source.SourceFunction
import scala.util.Random
class MySensorSource extends SourceFunction[SensorReading]{
//flag：表示数据源是否还在正常运行
var runing: Boolean = true
override def cancel(): Unit = {
runing=false
}
override def run(ctx: SourceFunction.SourceContext[SensorReading]): Unit = {
//初始化一个随机数发生器
val rand = new Random()
//初始化定义一组传感器温度数据
var curTemp=1.to(10).map(
i=>("sensor_"+i,65+rand.nextGaussian()*20)
)
while(runing){
//在前一次温度的基础上更新温度值
curTemp = curTemp.map(
t=>(t._1,t._2*rand.nextGaussian())
)
//获取当前时间戳
val curTime = System.currentTimeMillis();
curTemp.foreach(
t=>ctx.collect(SensorReading(t._1,curTime,t._2))
)
//设置时间间隔
Thread.sleep(500)
}
}
}
运行调用代码，结果控制台一直在输出不断随机生成的数据，如下：
