Apache Flink 入门

零、概述

Apache Flink 是一个高性能的开源分布式流处理框架，专注于实时数据流的处理。

它设计用于处理无界和有界数据流，在内存级速度下提供高效的有状态计算。

Flink 凭借其独特的Checkpoint机制和Exactly-Once语义，确保数据处理的准确性和一致性，同时支持高吞吐量和低延迟。

通过灵活的窗口操作和丰富的状态管理功能，Flink 能够应对复杂的实时数据处理需求，是大数据处理领域的重要技术之一。

其强大的DataStream API和Table API为开发者提供了高效、简洁的数据处理手段。

一、添加依赖 pom.xml

<?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>com.xch</groupId><artifactId>java-flink</artifactId><version>1.0-SNAPSHOT</version><properties><encoding>UTF-8</encoding><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target><java.version>1.8</java.version><flink.version>1.12.2</flink.version></properties><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.12</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.12</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java-bridge_2.12</artifactId><version>${flink.version}</version></dependency></dependencies></project>

二、`map()` `filter()` `flatMap()`方法示例

2.1 map()方法示例

简单处理，和java8的stream的map()类似，不过只能进行简单的处理，返回：数组元素自身的和

public static List<Integer> mapDemo(DataSource<Integer> dataSteam) throws Exception {return dataSteam.map(x -> x + x).collect();
}

2.2 filter()方法示例

过滤方法，返回偶数，

public static List<Integer> filterDemo(DataSource<Integer> dataSteam) throws Exception {return dataSteam.filter(x -> x % 2 == 0).collect();
}

2.3 flatMap()方法示例

flatMap方法可以处理复杂、定制化的逻辑，返回元素的类型也可以是复杂的；

第一个简单处理的示例

public static List<Object> flatMapDemo(DataSource<Integer> dataSteam) throws Exception {return dataSteam.flatMap(new FlatMapFunction<Integer, Object>() {@Overridepublic void flatMap(Integer integer, Collector<Object> collector) throws Exception {collector.collect(integer);collector.collect(integer * integer);}}).collect();
}

第二个复杂的示例

public static List<Map<Integer, Object>> flatMapDemo1(DataSource<Integer> dataSteam) throws Exception {return dataSteam.flatMap(new FlatMapFunction<Integer, Map<Integer, Object>>() {@Overridepublic void flatMap(Integer integer, Collector<Map<Integer, Object>> collector) throws Exception{Map<Integer, Object> hashMap = new HashMap<>();hashMap.put(integer, integer * integer);collector.collect(hashMap);}}).collect();
}

2.4 示例演示

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.util.Collector;import java.util.HashMap;
import java.util.List;
import java.util.Map;public class FlinkDemo {public static void main(String[] args) throws Exception {ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();DataSource<Integer> dataSteam = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);System.out.println("mapDemo:" + mapDemo(dataSteam));System.out.println("filterDemo:" + filterDemo(dataSteam));System.out.println("flatMapDemo:" + flatMapDemo(dataSteam));System.out.println("flatMapDemo1:" + flatMapDemo1(dataSteam));}
}