Flink 流处理API_大数据系统

Flink 流处理API

Environment getExecutionEnvironment

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

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

createLocalEnvironment

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

LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);

createRemoteEnvironment

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

StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("jobmanage-hostname", 6123,"YOURPATH//WordCount.jar");

Source 从集合读取数据

public class SourceTest1_Collection {
	public static void main(String[] args) throws Exception{ 
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();	
	// 1.Source:从集合读取数据
	DataStream sensorDataStream = env.fromCollection( 
		Arrays.asList(	
			new SensorReading("sensor_1", 1547718199L, 35.8), 
			new SensorReading("sensor_6", 1547718201L, 15.4), 
			new SensorReading("sensor_7", 1547718202L, 6.7), 
			new SensorReading("sensor_10", 1547718205L, 38.1)
		)
	);	
	// 2.打印
	sensorDataStream.print();
	// 3.执行
	env.execute();
	}
}

从文件读取数据

DataStream dataStream = env.readTextFile("YOUR_FILE_PATH ");

以 kafka 消息队列的数据作为来源

需要引入 kafka 连接器的依赖：pom.xml


	org.apache.flink
	flink-connector-kafka-0.11_2.12
	1.10.1

具体代码如下：

// kafka 配置项
Properties 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");
// 从 kafka 读取数据
DataStream dataStream = env.addSource( new FlinkKafkaConsumer011("sensor", new SimpleStringSchema(), properties));

自定义 Source

除了以上的 source 数据来源，我们还可以自定义 source。需要做的，只是传入一个 SourceFunction 就可以。具体调用如下：

DataStream dataStream = env.addSource( new MySensor());

我们希望可以随机生成传感器数据， MySensorSource 具体的代码实现如下：

public static class MySensor implements SourceFunction{
	private boolean running = true;
	public void run(SourceContext ctx) throws Exception { 
		Random random = new Random();
		HashMap sensorTempMap = new HashMap();
		for( int i = 0; i < 10; i++ ){
			sensorTempMap.put("sensor_" + (i + 1), 60 + random.nextGaussian() * 20);
		}
		while (running) {
			for( String sensorId: sensorTempMap.keySet() ){
				Double newTemp = sensorTempMap.get(sensorId) + random.nextGaussian(); 
				sensorTempMap.put(sensorId, newTemp);
				ctx.collect( new SensorReading(sensorId, System.currentTimeMillis(),newTemp));
			}
			Thread.sleep(1000L);
		}
	}
	public void cancel() {
		this.running = false;
	}
}

Transform转换算子 map

最基本的转换功能

DataStream mapStram = dataStream.map(new MapFunction() {
	public Integer map(String value) throws Exception {
		return value.length();
	}
});

flatMap

打散，拆分，输出多个数据

DataStream flatMapStream = dataStream.flatMap(new FlatMapFunction() {
	public void flatMap(String value, Collector out) throws Exception { 
		String[] fields = value.split(",");
		for( String field: fields )
		out.collect(field);
	}
});

Filter

按照某种条件过滤筛选，可能会不输出

DataStream filterStream = dataStream.filter(new FilterFunction()
{
	public boolean filter(String value) throws Exception {
		return value == 1;
	}
});

KeyBy

DataStream → KeyedStream：逻辑地将一个流拆分成不相交的分区，每个分区包含具有相同 key 的元素，在内部以 hash 的形式实现的。

滚动聚合算子（Rolling Aggregation）

这些算子可以针对 KeyedStream 的每一个支流做聚合。

sum()min()（只更新选择字段的内容）max()（只更新选择字段的内容）minBy()（更新选择字段的整条内容）maxBy()（更新选择字段的整条内容） Reduce

KeyedStream → DataStream：一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。

DataStream inputStream = env.readTextFile("sensor.txt");
// 转换成 SensorReading 类型
DataStream dataStream = inputStream.map(new MapFunction() {
	public SensorReading map(String value) throws Exception { 
		String[] fileds = value.split(",");
		return new SensorReading(fileds[0], new Long(fileds[1]), new Double(fileds[2]));
	}
});
// 分组
KeyedStream keyedStream = dataStream.keyBy("id");
// reduce 聚合，取最小的温度值，并输出当前的时间戳
DataStream reduceStream = keyedStream.reduce(new ReduceFunction() {
	@Override
	public SensorReading reduce(SensorReading value1, SensorReading value2)throws Exception {
		return new SensorReading( value1.getId(), value2.getTimestamp(),Math.min(value1.getTemperature(), value2.getTemperature()));
	}
});

Split 和 Select Split

DataStream → SplitStream：根据某些特征把一个 DataStream 拆分成两个或者多个 DataStream。

Select

SplitStream→DataStream：从一个 SplitStream 中获取一个或者多个DataStream。

需求：传感器数据按照温度高低（以 30 度为界），拆分成两个流。

SplitStream splitStream = dataStream.split(new OutputSelector() {
	@Override
	public Iterable select(SensorReading value) {
		return (value.getTemperature() > 30) ? Collections.singletonList("high") : Collections.singletonList("low");
	}
});
DataStream highTempStream = splitStream.select("high"); 
DataStream lowTempStream = splitStream.select("low"); 
DataStream allTempStream = splitStream.select("high", "low");

Connect 和 CoMap

DataStream,DataStream → ConnectedStreams：连接两个保持他们类型的数据流，两个数据流被 Connect 之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。

CoMap,CoFlatMap

ConnectedStreams → DataStream ：作用于 ConnectedStreams 上，功能与 map 和 flatMap 一样，对 ConnectedStreams 中的每一个 Stream 分别进行 map 和 flatMap 处理。

//合流 connect
DataStream> warningStream = highTempStream.map(new MapFunction>() {
	@Override
	public Tuple2 map(SensorReading value) throws Exception {
		return new Tuple2<>(value.getId(), value.getTemperature());
	}	
});
ConnectedStreams, SensorReading> connectedStreams = warningStream.connect(lowTempStream);
DataStream

Flink 流处理API

大数据系统相关栏目本月热门文章