.png)
.png)
Apache Flink 是一个强大的流处理框架，广泛应用于实时数据分析和处理。本文将带你从零开始，
使用 Flink 开发一个简单的流处理程序，处理车辆定位物联网场景中的 GPS 信息。我们将从安装和配置环境开始，逐步构建并运行这个程序。
1. 环境准备，jdk+maven+idea 此步骤略
2. 创建 Maven 项目
目录结构如下
修改 pom.xml 文件，添加 Flink 依赖：
<?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-st1</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.14.2</version>
</dependency>
<!-- 如果是Maven项目，请确保pom.xml中有以下依赖 -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.12</artifactId>
<version>1.14.2</version> <!-- 确保版本与你的环境匹配 -->
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.12</artifactId>
<version>1.14.2</version> <!-- 确保版本与你的环境匹配 -->
</dependency>
</dependencies>
</project>3. 定义 GPS 数据结构
首先，我们需要定义一个类来表示 GPS 数据。创建一个名为 GpsData.java 的文件，内容如下
package org.example.vehiclegps;
public class GpsData {
private String vehicleId;
private double lat;
private double lon;
private long gpsTime;
public GpsData(String vehicleId, double lat, double lon, long timestamp) {
this.vehicleId = vehicleId;
this.lat = lat;
this.lon = lon;
this.gpsTime = timestamp;
}
public String getVehicleId() {
return vehicleId;
}
public double getLatitude() {
return lat;
}
public double getLongitude() {
return lon;
}
public long getTimestamp() {
return gpsTime;
}
@Override
public String toString() {
return "GpsData{" +
"vehicleId='" + vehicleId + '\'' +
", lat=" + lat +
", lon=" + lon +
", gpsTime=" + gpsTime +
'}';
}
}
4. 创建 Socket 服务器
接下来，创建一个 Socket 服务器，用于生成并发送模拟的 GPS 数据。创建一个名为 RandomGpsSocketServer.java 的文件
package org.example.vehiclegps;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
public class FirstFlinkJob {
public static void main(String[] args) throws Exception {
// 设置执行环境
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 创建一个数据流，这里以socket文本流为例
DataStream<String> text = env.socketTextStream("localhost", 9999);
// 处理数据流
DataStream<GpsData> gpsDataStream = text
.flatMap(new GpsDataParser());
// 打印结果至控制台
gpsDataStream.print();
// 开始执行
env.execute("Vehicle GPS Data Processing");
}
// 实现FlatMapFunction接口，用来解析GPS数据
public static final class GpsDataParser implements FlatMapFunction<String, GpsData> {
@Override
public void flatMap(String value, Collector<GpsData> out) {
try {
String[] parts = value.split(",");
if (parts.length == 4) {
String vehicleId = parts[0].trim().replace("GpsData{vehicleId='", "").replace("'", "");
double latitude = Double.parseDouble(parts[1].trim().replace("lat=", "").replace(",", ""));
double longitude = Double.parseDouble(parts[2].trim().replace("lon=", "").replace(",", ""));
long timestamp = Long.parseLong(parts[3].trim().replace("gpsTime=", "").replace("}", ""));
out.collect(new GpsData(vehicleId, latitude, longitude, timestamp));
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
5. 编写 Flink 作业
现在，我们编写 Flink 作业来处理从 Socket 服务器接收到的 GPS 数据。FirstFlinkJob.java
package org.example.vehiclegps;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Random;
public class RandomGpsSocketServer {
private static final int PORT = 9999;
private static final String[] VEHICLE_IDS = {"V1001", "V1002", "V1003", "V1004", "V1005"};
public static void main(String[] args) {
try (ServerSocket serverSocket = new ServerSocket(PORT)) {
System.out.println("Socket服务器已启动，等待客户端连接...");
try (Socket clientSocket = serverSocket.accept()) {
System.out.println("客户端已连接");
OutputStream outputStream = clientSocket.getOutputStream();
PrintWriter out = new PrintWriter(outputStream, true);
// 模拟持续发送随机生成的GPS数据
Random random = new Random();
while (true) {
GpsData gpsData = generateRandomGpsData(random);
out.println(gpsData.toString());
System.out.println("发送: " + gpsData.toString());
Thread.sleep(1000); // 每秒发送一次
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
private static GpsData generateRandomGpsData(Random random) {
String vehicleId = VEHICLE_IDS[random.nextInt(VEHICLE_IDS.length)];
double latitude = -90 + (180 * random.nextDouble());
double longitude = -180 + (360 * random.nextDouble());
long timestamp = System.currentTimeMillis();
return new GpsData(vehicleId, latitude, longitude, timestamp);
}
}
6. Flink 作业 运行结果
模拟车载设备上报定位数据
flink 接收和处理结构
常见错误
No ExecutorFactory found to execute the application. 一般为maven依赖配置的不对或者冲突
Exception in thread "main" java.lang.IllegalStateException: No ExecutorFactory found to execute the application.
at org.apache.flink.core.execution.DefaultExecutorServiceLoader.getExecutorFactory(DefaultExecutorServiceLoader.java:88)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2033)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1931)
at org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:69)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1917)
at com.e6yun.epaas.admin.FirstFlinkProgram.main(FirstFlinkProgram.java:18)
使用上海交大的镜像库
评论区