Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 5e8e5f6a by 吴延飞
Options

整理目录结构，删除无用文件夹和文件

parent 15a5fcc3
Showing with 6 additions and 1242 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/kafka/CycleInfo.java deleted 100644 → 0
package com.zhht.irn.entity.kafka;
import java.util.Date;
import java.util.List;
/**
* 周期数据（信控数据）
* 每个路口的信号控制等每完成一个周期，就发送该周期的信息到实时数据流中。
* 数据结构参考最新的实时数据文档 --> 周期数据（信控）
* 全息路口实时数据文档v1.1 (2022.11.1）》
* https://www.yuque.com/docs/share/9cd59d71-6165-46da-ae43-bc5388149672?#
*
*/
public class CycleInfo {
// ID
private Integer id;
// 路口编号
private String crossCode;
// 信控机编号
private String signalCode;
// 开始模式编码
private String beginControlModelCode;
// 结束模式编码
private String endControlModelCode;
// 开始方案ID
private Integer beginPlanId;
// 结束方案ID
private Integer endPlanId;
// 周期开始时间
private Date beginDateTime;
// 周期结束时间
private Date endDateTime;
// 周期时长，单位 秒
private Double duration;
// 周期顺序
private Integer cycleOrder;
// 相位列表集合
private List<Stage> stageList;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/kafka/Stage.java deleted 100644 → 0
package com.zhht.irn.entity.kafka;
import java.util.Date;
public class Stage {
// ID
private Integer id;
// 相位值
private String phaseValue;
// 相位开始时间
private Date beginDateTime;
// 相位结束时间
private Date endDateTime;
// 开始模式编码
private String beginControlModelCode;
// 结束模式编码
private String endControlModelCode;
// 开始方案ID
private Integer beginPlanId;
// 结束方案ID
private Integer endPlanId;
// 周期时长，单位 秒
private Integer duration;
// 绿灯时长，单位 秒
private Integer green;
// 黄灯时长，单位 秒
private Integer yellow;
// 全红时长，单位 秒
private Integer allRed;
// 周期顺序
private Integer phaseOrder;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/mysql/TripsMetric.java deleted 100644 → 0
package com.zhht.irn.entity.mysql;
import java.util.Date;
/**
* 旅行信息的准实时指标对应
*
* 纬度：车辆、路口
* 指标：平均速度、最高速度、最低速度、停车次数、停车总时长
*
* @author 吴延飞
* @date 2022-11-02 15:53:00
*/
public class TripsMetric {
// 旅行记录Id
private String recordId;
// 捕获时间
private Date captureTime;
// 丢失时间
private Date lostTime;
// 最高时速
private Double maxSpeed;
// 最低时速
private Double minSpeed;
// 平均时速
private Double avgSpeed;
// 停车总时长
private Integer stopDuration;
// 停车总次数
private Integer stopCount;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/FlinkKafka2MysqlApp.java deleted 100644 → 0
package com.zhht.irn.flow;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.sink.LXBSink;
import org.apache.flink.api.common.functions.MapFunction;
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 java.util.List;
/**
* 两阶段提交参考文档：
* https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka
*/
public class FlinkKafka2MysqlApp {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// DataStream<String> stream = FlinkUtils.createKafkaStreamV2(args, SimpleStringSchema.class);
// 目前假设从Kafka接入的数据为json格式的字符串
// Cycle Topic 的消息格式如下：
// {"Id":"","CrossCode":"","BeginTime":"","EndTime":"","Duration":"","CycleOrder":"",
// "detail":[{"PhaseValue":"101","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"102","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"103","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"104","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""}]
// 造一条数据
DataStream<String> stream = env.fromElements("{\"id\":\"1\",\"crossCode\":\"0001\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"300\",\"cycleOrder\":\"1\",\"phaseList\":[{\"phaseValue\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:01:00\",\"duration\":\"60\",\"green\":\"30\",\"yellow\":\"3\",\"allRed\":\"0\"},{\"phaseValue\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\"}]}");
DataStream<Tuple2<String, Double>> lxbStream = stream.map(new MapFunction<String, Tuple2<String, Double>>() {
@Override
public Tuple2<String, Double> map(String s) throws Exception {
//启动损失时间 = 2 ，从配置文件读取，正常情况下都是2，不排除个别城市有差异
//清场损失时间 = 黄灯时长 + 全红时长 - 2
//绿信比 = 每个相位的绿灯时长累加 - 启动损失时间 - 清场损失时间
Cycle cycle = JSON.parseObject(s, Cycle.class);
Long allValidGreen = 0L;
List<Stage> phaseList = cycle.getStageList();
for (Stage phase : phaseList) {
// 绿灯时间
long green = phase.getGreen();
// 黄灯时间
long yellow = phase.getYellow();
// 全红时间
long allRed = phase.getAllRed();
// 启动损失时间取一般定义：2s
long startLossTime = 2L;
//清场损失时间
long clearLossTime = yellow + allRed - 2;
allValidGreen = allValidGreen + green - startLossTime - clearLossTime;
}
// 绿信比
Double lxb = allValidGreen.doubleValue() / cycle.getDuration();
return Tuple2.of(cycle.getCrossCode(), lxb);
}
});
lxbStream.addSink(new LXBSink());
env.execute();
// FlinkUtils.env.execute();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/LaneMetricJob.java deleted 100644 → 0
package com.zhht.irn.flow;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.TravelEvent;
import com.zhht.irn.functions.CalLaneMetricProcessFunction2;
import com.zhht.irn.source.CycleMockSource;
import com.zhht.irn.source.TravelEventMockSource;
import com.zhht.irn.trigger.CycleTimeTrigger;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.window.CycleWindowAssigner;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.*;
/**
* 车道级指标的计算任务
* 按照周期进行实时计算车道级的指标
* 计算的指标有：
* 最大排队长度、有效绿灯时间、通行能力、交通流量、饱和度、剩余承载力、识别区空间占有率
*
* @author 吴延飞
* @date 2022-11-13 13:55:00
*/
public class LaneMetricJob {
private static final Logger logger = LoggerFactory.getLogger(LaneMetricJob.class);
public static void main(String[] args) throws Exception {
// 创建实时数据流上下文
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 旅行事件数据流的 watermark 生成逻辑
WatermarkStrategy<TravelEvent> travelEventWatermarkStrategy = WatermarkStrategy.<TravelEvent>forBoundedOutOfOrderness(Duration.ofSeconds(30))
.withTimestampAssigner((travelEvent, timestamp) -> DateUtils.dateToStamp(travelEvent.getEventTime()));
// 周期结束数据流的 cycle 生成逻辑
WatermarkStrategy<Cycle> cycleWatermarkStrategy = WatermarkStrategy.<Cycle>forBoundedOutOfOrderness(Duration.ofSeconds(10))
.withTimestampAssigner((cycle, timeStamp) -> DateUtils.dateToStamp(cycle.getBeginDateTime()));
// 获取实时旅行事件数据流
// DataStream<TravelEvent> eventStream = FlinkUtils.createKafkaStream(args, env)
DataStream<TravelEvent> eventStream = env.addSource(new TravelEventMockSource())
// 格式解析，将Sting转化为实体类, 存在无法直接从String转为Entity的情况，需字段单独解析 @TODO
.map((record) -> {
System.out.println(record);
TravelEvent travelEvent = JSON.parseObject(record, TravelEvent.class);
System.out.println(travelEvent);
return travelEvent;
})
// 给数据流添加水印
.assignTimestampsAndWatermarks(travelEventWatermarkStrategy);
// 获取实时周期结束数据流
// DataStream<Cycle> cycleStream = FlinkUtils.createKafkaStream(args, env)
DataStream<Cycle> cycleStream = env.addSource(new CycleMockSource())
// 格式解析，将Sting转化为实体类, 存在无法直接从String转为Entity的情况，需字段单独解析 @TODO
.map((record) -> {
System.out.println(record);
Cycle cycle = JSON.parseObject(record, Cycle.class);
return cycle;
})
// 给数据流添加水印
.assignTimestampsAndWatermarks(cycleWatermarkStrategy);
// 通过join操作，获取一个Tuple<List<TravelEvent>, Cycle> 类型的数据流
// DataStream<Tuple2<Cycle, TravelEvent>> joinStream = cycleStream.coGroup(eventStream)
// // 根据路口分组
// .where(cycle -> cycle.getCrossCode())
// .equalTo(travelEvent -> travelEvent.getCrossId())
// // 根据周期来进行窗口划分
// .window(new CycleWindowAssigner())
// .trigger(new CycleTimeTrigger())
// .allowedLateness(Time.minutes(10))
// //
// .apply(new CoGroupFunction<Cycle, TravelEvent, Tuple2<Cycle, TravelEvent>>() {
// @Override
// public void coGroup(Iterable<Cycle> iterable, Iterable<TravelEvent> iterable1, Collector<Tuple2<Cycle, TravelEvent>> collector) throws Exception {
// collector.collect(Tuple2.of(iterable.iterator().next(), iterable1.iterator().next()));
// }
// });
ConnectedStreams<Cycle, TravelEvent> cycleTravelEventConnectedStreams = cycleStream.connect(eventStream).keyBy(cycle -> cycle.getCrossCode(), travelEvent -> travelEvent.getCrossId());
cycleTravelEventConnectedStreams.process(new CalLaneMetricProcessFunction2())
.print();
env.execute();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/LXBSink.java deleted 100644 → 0
package com.zhht.irn.sink;
import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.PreparedStatement;
/**
* domain traffic
*/
public class LXBSink extends RichSinkFunction<Tuple2<String, Double>> {
Connection connection;
PreparedStatement insertPstmt;
PreparedStatement updatePstmt;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// path暂定为固定值
String path = "E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
connection = MySQLUtils.getConnection(path);
insertPstmt = connection.prepareStatement("insert into wyf_test(cross_code,lxb) values (?,?)");
updatePstmt = connection.prepareStatement("update wyf_test set lxb=? where cross_code=?");
}
@Override
public void close() throws Exception {
super.close();
if(insertPstmt != null) insertPstmt.close();
if(updatePstmt != null) updatePstmt.close();
if(connection != null) connection.close();
}
/**
* 来一条数据就执行一次
*
* 1000w的数据 1000w次
*/
@Override
public void invoke(Tuple2<String, Double> value, Context context) throws Exception {
System.out.println("=====invoke======" + value.f0 + "-->" + value.f1);
updatePstmt.setDouble(1, value.f1);
updatePstmt.setString(2 , value.f0);
updatePstmt.execute();
if(updatePstmt.getUpdateCount() == 0) {
insertPstmt.setString(1, value.f0);
insertPstmt.setDouble(2, value.f1);
insertPstmt.execute();
}
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/PKRedisSink.java deleted 100644 → 0
package com.zhht.irn.sink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
/**
* 在生产环境中，
*
* 软件的版本做了升级
* 代码有了很大变化
*
* ==> diff
*
*
*/
public class PKRedisSink implements RedisMapper<Tuple2<String, Double>> {
@Override
public RedisCommandDescription getCommandDescription() {
return new RedisCommandDescription(RedisCommand.HSET, "pk-traffic");
}
@Override
public String getKeyFromData(Tuple2<String, Double> data) {
return data.f0;
}
@Override
public String getValueFromData(Tuple2<String, Double> data) {
return data.f1 +"";
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/SinkApp.java deleted 100644 → 0
package com.zhht.irn.sink;
import com.zhht.irn.transformation.Access;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
public class SinkApp {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
toMySQL(env);
env.execute("SinkApp");
}
public static void toMySQL(StreamExecutionEnvironment env) {
DataStreamSource<String> source = env.readTextFile("data/access.log");
SingleOutputStreamOperator<Access> mapStream = source.map(new MapFunction<String, Access>() {
@Override
public Access map(String value) throws Exception {
String[] splits = value.split(",");
Long time = Long.parseLong(splits[0].trim());
String domain = splits[1].trim();
Double traffic = Double.parseDouble(splits[2].trim());
return new Access(time, domain, traffic);
}
});
SingleOutputStreamOperator<Access> result = mapStream.keyBy(new KeySelector<Access, String>() {
@Override
public String getKey(Access value) throws Exception {
return value.getDomain();
}
}).sum("traffic");
result.print();
FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").build();
result.map(new MapFunction<Access, Tuple2<String, Double>>() {
@Override
public Tuple2<String, Double> map(Access value) throws Exception {
return Tuple2.of(value.getDomain(), value.getTraffic());
}
}) // .addSink(new PKMySQLSink());
.addSink(new RedisSink<Tuple2<String, Double>>(conf, new PKRedisSink()));
}
public static void print(StreamExecutionEnvironment env) {
DataStreamSource<String> source = env.socketTextStream("localhost", 9527);
System.out.println("source:" + source.getParallelism());
source.print().setParallelism(2);
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TravelMetricSink.java
package com.zhht.irn.sink; package com.zhht.irn.sink;
import com.zhht.irn.entity.Travel; import com.zhht.irn.entity.Travel;
import com.zhht.irn.entity.mysql.TripsMetric;
import com.zhht.irn.utils.DateUtils; import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.MySQLUtils; import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/AccessSource.java deleted 100644 → 0
package com.zhht.irn.source;
import com.zhht.irn.transformation.Access;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.util.Random;
public class AccessSource implements SourceFunction<Access> {
boolean running = true;
@Override
public void run(SourceContext<Access> ctx) throws Exception {
String[] domains = {"imooc.com", "a.com","b.com"};
Random random = new Random();
while (running) {
for (int i = 0; i < 10 ; i++) {
Access access = new Access();
access.setTime(1234567L);
access.setDomain(domains[random.nextInt(domains.length)]);
access.setTraffic(random.nextDouble() + 1000);
ctx.collect(access);
}
Thread.sleep(5000);
}
}
@Override
public void cancel() {
running = false;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/AccessSourceV2.java deleted 100644 → 0
package com.zhht.irn.source;
import com.zhht.irn.transformation.Access;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import java.util.Random;
public class AccessSourceV2 implements ParallelSourceFunction<Access> {
boolean running = true;
@Override
public void run(SourceContext<Access> ctx) throws Exception {
String[] domains = {"imooc.com", "a.com","b.com"};
Random random = new Random();
while (running) {
for (int i = 0; i < 10 ; i++) {
Access access = new Access();
access.setTime(1234567L);
access.setDomain(domains[random.nextInt(domains.length)]);
access.setTraffic(random.nextDouble() + 1000);
ctx.collect(access);
}
Thread.sleep(5000);
}
}
@Override
public void cancel() {
running = false;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/SourceApp.java deleted 100644 → 0
package com.zhht.irn.source;
import com.zhht.irn.transformation.Access;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.NumberSequenceIterator;
import java.util.Properties;
public class SourceApp {
public static void main(String[] args) throws Exception {
// 创建上下文
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// test01(env);
// test02(env);
// test03(env);
// test04(env);
test05(env);
env.execute("SourceApp");
}
public static void test05(StreamExecutionEnvironment env ) {
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "ruozedata001:9092,ruozedata001:9093,ruozedata001:9094");
properties.setProperty("group.id", "test");
DataStream<String> stream = env
.addSource(new FlinkKafkaConsumer<>("flinktopic", new SimpleStringSchema(), properties));
System.out.println(stream.getParallelism());
stream.print();
}
public static void test04(StreamExecutionEnvironment env ) {
DataStreamSource<Student> source = env.addSource(new StudentSource()).setParallelism(3);
System.out.println(source.getParallelism());
source.print();
}
public static void test03(StreamExecutionEnvironment env ) {
// DataStreamSource<Access> source = env.addSource(new AccessSource())
// .setParallelism(2);
DataStreamSource<Access> source = env.addSource(new AccessSourceV2()).setParallelism(3);
System.out.println(source.getParallelism());
source.print();
}
public static void test02(StreamExecutionEnvironment env ){
env.setParallelism(5); // 对于env设置的并行度 是一个全局的概念
DataStreamSource<Long> source = env.fromParallelCollection(
new NumberSequenceIterator(1, 10), Long.class
);//.setParallelism(4);
System.out.println("source:" + source.getParallelism());
SingleOutputStreamOperator<Long> filterStream = source.filter(new FilterFunction<Long>() {
@Override
public boolean filter(Long value) throws Exception {
return value >= 5;
}
}).setParallelism(3); // 对于算子层面的并行度，如果全局设置，以本算子的并行度为准
System.out.println("filterStream:" + filterStream.getParallelism());
filterStream.print();
}
public static void test01(StreamExecutionEnvironment env ){
env.setParallelism(5);
// StreamExecutionEnvironment.createLocalEnvironment();
// StreamExecutionEnvironment.createLocalEnvironment(3);
// StreamExecutionEnvironment.createLocalEnvironment(new Configuration());
// StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
// StreamExecutionEnvironment.createRemoteEnvironment()
DataStreamSource<String> source = env.socketTextStream("localhost", 9527);
System.out.println("source...." + source.getParallelism()); // ? 1
// 接收socket过来的数据，一行一个单词， 把pk的过滤掉
SingleOutputStreamOperator<String> filterStream = source.filter(new FilterFunction<String>() {
@Override
public boolean filter(String value) throws Exception {
return !"pk".equals(value);
}
}).setParallelism(6);
System.out.println("filter...." + filterStream.getParallelism());
filterStream.print();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/Student.java deleted 100644 → 0
package com.zhht.irn.source;
public class Student {
private int id;
private String name;
private int age;
public Student() {
}
@Override
public String toString() {
return "Student{" +
"id=" + id +
", name='" + name + '\'' +
", age=" + age +
'}';
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
public Student(int id, String name, int age) {
this.id = id;
this.name = name;
this.age = age;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/StudentSource.java deleted 100644 → 0
package com.zhht.irn.source;
import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
public class StudentSource extends RichSourceFunction<Student> {
Connection connection;
PreparedStatement psmt;
@Override
public void open(Configuration parameters) throws Exception {
connection = MySQLUtils.getConnection(parameters.getString("path", ""));
psmt = connection.prepareStatement("select * from student");
}
@Override
public void close() throws Exception {
MySQLUtils.close(connection, psmt);
}
@Override
public void run(SourceContext<Student> ctx) throws Exception {
ResultSet rs = psmt.executeQuery();
while (rs.next()) {
int id = rs.getInt("id");
String name = rs.getString("name");
int age = rs.getInt("age");
ctx.collect(new Student(id, name, age));
}
}
@Override
public void cancel() {
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/TravelMockSource.java
...@@ -41,25 +41,26 @@ public class TravelMockSource extends RichSourceFunction<String> { ...@@ -41,25 +41,26 @@ public class TravelMockSource extends RichSourceFunction<String> {
"]}"; "]}";
// 设置参数 // 设置参数
long current = 1668391200000L; // 固定为 2022-11-14 10:00::00
Map<String, Object> params = new HashMap<>(); Map<String, Object> params = new HashMap<>();
params.put("id", 1); params.put("id", 1);
params.put("crossId", "13070200137"); params.put("crossId", "13070200137");
params.put("recordId", "00001"); params.put("recordId", "00001");
params.put("recordTime", 1667121436000L); params.put("recordTime", current);
params.put("carId", "10"); params.put("carId", "10");
params.put("carColor", ""); params.put("carColor", "");
params.put("plate", ""); params.put("plate", "");
params.put("plateColor", ""); params.put("plateColor", "");
params.put("carType", "0"); params.put("carType", "0");
params.put("firstLineId", ""); params.put("firstLineId", "");
params.put("firstTime", 1667121422000L); params.put("firstTime", current - 90000);
params.put("firstSpeed", 56.8609); params.put("firstSpeed", 56.8609);
params.put("arrivedLineId", "11"); params.put("arrivedLineId", "11");
params.put("arrivedTime", 1667121426000L); params.put("arrivedTime", current - 80000);
params.put("arrivedSpeed", 58.5416); params.put("arrivedSpeed", 58.5416);
params.put("inCrossLineId", "12"); params.put("inCrossLineId", "12");
params.put("inCrossTime", 1667121434000L); params.put("inCrossTime", current - 70000);
params.put("inSpeed", 58.7933); params.put("inSpeed", 58.7933);
params.put("outCrossLineId", ""); params.put("outCrossLineId", "");
...@@ -71,7 +72,7 @@ public class TravelMockSource extends RichSourceFunction<String> { ...@@ -71,7 +72,7 @@ public class TravelMockSource extends RichSourceFunction<String> {
params.put("awaySpeed", 0); params.put("awaySpeed", 0);
params.put("lastLineId", ""); params.put("lastLineId", "");
params.put("lastTime", 1667121436000L); params.put("lastTime", current);
params.put("lastSpeed", 58.6742); params.put("lastSpeed", 58.6742);
params.put("crossingTime", 0); params.put("crossingTime", 0);
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/transformation/Access.java deleted 100644 → 0
package com.zhht.irn.transformation;
public class Access {
private Long time;
private String domain;
private Double traffic;
public Access() {
}
public Access(Long time, String domain, Double traffic) {
this.time = time;
this.domain = domain;
this.traffic = traffic;
}
@Override
public String toString() {
return "Access{" +
"time=" + time +
", domain='" + domain + '\'' +
", traffic=" + traffic +
'}';
}
public Long getTime() {
return time;
}
public void setTime(Long time) {
this.time = time;
}
public String getDomain() {
return domain;
}
public void setDomain(String domain) {
this.domain = domain;
}
public Double getTraffic() {
return traffic;
}
public void setTraffic(Double traffic) {
this.traffic = traffic;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/transformation/PKMapFunction.java deleted 100644 → 0
package com.zhht.irn.transformation;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
public class PKMapFunction extends RichMapFunction<String,Access> {
/**
* 初始化操作
* Connection
*/
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
System.out.println("~~~~open~~~~");
}
/**
* 清理操作
*/
@Override
public void close() throws Exception {
super.close();
}
@Override
public RuntimeContext getRuntimeContext() {
return super.getRuntimeContext();
}
/**
* 每条数据执行一次
*/
@Override
public Access map(String value) throws Exception {
System.out.println("=====map=====");
String[] splits = value.split(",");
Long time = Long.parseLong(splits[0].trim());
String domain = splits[1].trim();
Double traffic = Double.parseDouble(splits[2].trim());
return new Access(time, domain, traffic);
}
}
\ No newline at end of file
realtime/hologram-streaming/src/main/java/com/zhht/irn/transformation/TransformationApp.java deleted 100644 → 0
package com.zhht.irn.transformation;
import com.zhht.irn.source.AccessSource;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;
import java.util.ArrayList;
public class TransformationApp {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// map(env);
// filter(env);
// flatMap(env);
// keyBy(env);
// reduce(env);
// richMap(env);
// union(env);
// connect(env);
// coMap(env);
coFlatMap(env);
env.execute("SourceApp");
}
public static void coFlatMap(StreamExecutionEnvironment env) {
DataStreamSource<String> stream1 = env.fromElements("a b c", "d e f");
DataStreamSource<String> stream2 = env.fromElements("1,2,3", "4,5,6");
stream1.connect(stream2)
.flatMap(new CoFlatMapFunction<String, String, String>() {
@Override
public void flatMap1(String value, Collector<String> out) throws Exception {
String[] splits = value.split(" ");
for(String split : splits) {
out.collect(split);
}
}
@Override
public void flatMap2(String value, Collector<String> out) throws Exception {
String[] splits = value.split(",");
for(String split : splits) {
out.collect(split);
}
}
}).print();
}
public static void coMap(StreamExecutionEnvironment env) {
DataStreamSource<String> stream1 = env.socketTextStream("localhost", 9527);
SingleOutputStreamOperator<Integer> stream2 = env.socketTextStream("localhost", 9528) // 数值类型
.map(new MapFunction<String, Integer>() {
@Override
public Integer map(String value) throws Exception {
return Integer.parseInt(value);
}
});
// 将2个流连接在一起
stream1.connect(stream2).map(new CoMapFunction<String, Integer, String>() {
// 处理第一个流的业务逻辑
@Override
public String map1(String value) throws Exception {
return value.toUpperCase();
}
// 处理第二个流的业务逻辑
@Override
public String map2(Integer value) throws Exception {
return value * 10 + "";
}
}).print();
}
/**
* union 多流合并 数据结构必须相同
* connect 双流 数据结构可以不同， 更加灵活
*/
public static void connect(StreamExecutionEnvironment env) {
DataStreamSource<Access> stream1 = env.addSource(new AccessSource());
DataStreamSource<Access> stream2 = env.addSource(new AccessSource());
SingleOutputStreamOperator<Tuple2<String, Access>> stream2new = stream2.map(new MapFunction<Access, Tuple2<String, Access>>() {
@Override
public Tuple2<String, Access> map(Access value) throws Exception {
return Tuple2.of("pk", value);
}
});
stream1.connect(stream2new).map(new CoMapFunction<Access, Tuple2<String,Access>, String>() {
@Override
public String map1(Access value) throws Exception {
return value.toString();
}
@Override
public String map2(Tuple2<String, Access> value) throws Exception {
return value.f0 + "==>" + value.f1.toString();
}
}).print();
// ConnectedStreams<Access, Access> connect = stream1.connect(stream2);
//
// connect.map(new CoMapFunction<Access, Access, Access>() {
// @Override
// public Access map1(Access value) throws Exception {
// return value;
// }
//
// @Override
// public Access map2(Access value) throws Exception {
// return value;
// }
// }).print();
}
public static void union(StreamExecutionEnvironment env) {
DataStreamSource<String> stream1 = env.socketTextStream("localhost", 9527);
DataStreamSource<String> stream2 = env.socketTextStream("localhost", 9528);
DataStream<String> union = stream1.union(stream2);
// stream1.union(stream2).print();
stream1.union(stream1).print();
}
public static void richMap(StreamExecutionEnvironment env) {
env.setParallelism(3);
DataStreamSource<String> source = env.readTextFile("data/access.log");
SingleOutputStreamOperator<Access> mapStream = source.map(new PKMapFunction());
mapStream.print();
}
/**
* wc： socket
*
* 进来的数据：pk,pk,flink pk,spark,spark
*
*
* wc需求分析：
* 1) 读进来数据
* 2) 按照指定分隔符进行拆分 pk pk flink pk spark spark
* 3) 为每个单词赋上一个出现次数为1的值 (pk,1) (pk,1) ...
* 4) 按照单词进行keyBy
* 5) 分组求和
*
*/
public static void reduce(StreamExecutionEnvironment env) {
DataStreamSource<String> source = env.socketTextStream("localhost", 9527);
source.flatMap(new FlatMapFunction<String, String>() {
@Override
public void flatMap(String value, Collector<String> out) throws Exception {
String[] splits = value.split(",");
for(String word : splits) {
out.collect(word);
}
}
}).map(new MapFunction<String, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> map(String value) throws Exception {
return Tuple2.of(value, 1);
}
}).keyBy(x -> x.f0) // word相同的都会分到一个task中去执行
.reduce(new ReduceFunction<Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
return Tuple2.of(value1.f0, value1.f1 + value2.f1);
}
}).print();
}
/**
* 按照domain分组，求traffic和
*/
public static void keyBy(StreamExecutionEnvironment env) {
DataStreamSource<String> source = env.readTextFile("data/access.log");
SingleOutputStreamOperator<Access> mapStream = source.map(new MapFunction<String, Access>() {
@Override
public Access map(String value) throws Exception {
String[] splits = value.split(",");
Long time = Long.parseLong(splits[0].trim());
String domain = splits[1].trim();
Double traffic = Double.parseDouble(splits[2].trim());
return new Access(time, domain, traffic);
}
});
// mapStream.keyBy("domain").sum("traffic").print();
// mapStream.keyBy(new KeySelector<Access, String>() {
// @Override
// public String getKey(Access value) throws Exception {
// return value.getDomain();
// }
// }).sum("traffic").print();
KeyedStream<Access, String> keyedStream = mapStream.keyBy(x -> x.getDomain());
keyedStream.sum("traffic").print();
}
/**
* 进来是一行行的数据： pk,pk,flink pk,spark,spark
* 需求：
* 1) 把一行数据按照逗号进行分割
* 2) 过滤掉pk
*/
public static void flatMap(StreamExecutionEnvironment env) {
DataStreamSource<String> source = env.socketTextStream("localhost", 9527);
source.flatMap(new FlatMapFunction<String, String>() {
@Override
public void flatMap(String value, Collector<String> out) throws Exception {
String[] splits = value.split(",");
for(String split : splits) {
out.collect(split);
}
}
}).filter(new FilterFunction<String>() {
@Override
public boolean filter(String value) throws Exception {
return !"pk".equals(value);
}
}).print();
}
/**
* filter 就是对DataStream中的数据进行过滤操作
* 保留true
*/
public static void filter(StreamExecutionEnvironment env) {
DataStreamSource<String> source = env.readTextFile("data/access.log");
SingleOutputStreamOperator<Access> mapStream = source.map(new MapFunction<String, Access>() {
@Override
public Access map(String value) throws Exception {
String[] splits = value.split(",");
Long time = Long.parseLong(splits[0].trim());
String domain = splits[1].trim();
Double traffic = Double.parseDouble(splits[2].trim());
return new Access(time, domain, traffic);
}
});
SingleOutputStreamOperator<Access> filterStream = mapStream.filter(new FilterFunction<Access>() {
@Override
public boolean filter(Access value) throws Exception {
return value.getTraffic() > 4000;
}
});
filterStream.print();
}
/**
* 读进来的数据是一行行的，也字符串类型
*
* 每一行数据 ==> Access
*
* 将map算子对应的函数作用到DataStream，产生一个新的DataStream
*
* map会作用到已有的DataStream这个数据集中的每一个元素上
*
*/
public static void map(StreamExecutionEnvironment env) {
// DataStreamSource<String> source = env.readTextFile("data/access.log");
//
// SingleOutputStreamOperator<Access> mapStream = source.map(new MapFunction<String, Access>() {
// @Override
// public Access map(String value) throws Exception {
// String[] splits = value.split(",");
// Long time = Long.parseLong(splits[0].trim());
// String domain = splits[1].trim();
// Double traffic = Double.parseDouble(splits[2].trim());
//
// return new Access(time, domain, traffic);
// }
// });
//
// mapStream.print();
ArrayList<Integer> list = new ArrayList<>();
list.add(1); // map * 2 = 2
list.add(2); // map * 2 = 4
list.add(3); // map * 2 = 6
DataStreamSource<Integer> source = env.fromCollection(list);
source.map(new MapFunction<Integer, Integer>() {
@Override
public Integer map(Integer value) throws Exception {
return value * 2;
}
}).print();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/trigger/CycleTimeTrigger.java deleted 100644 → 0
package com.zhht.irn.trigger;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.TravelEvent;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
public class CycleTimeTrigger extends Trigger<CoGroupedStreams.TaggedUnion<Cycle, TravelEvent>, TimeWindow> {
@Override
public TriggerResult onElement(CoGroupedStreams.TaggedUnion<Cycle, TravelEvent> element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
// if the watermark is already past the window fire immediately
return TriggerResult.FIRE;
} else {
ctx.registerEventTimeTimer(window.maxTimestamp());
return TriggerResult.CONTINUE;
}
}
@Override
public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;
}
@Override
public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
ctx.deleteEventTimeTimer(window.maxTimestamp());
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/window/CycleWindowAssigner.java deleted 100644 → 0
package com.zhht.irn.window;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.TravelEvent;
import com.zhht.irn.trigger.CycleTimeTrigger;
import com.zhht.irn.utils.DateUtils;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import java.util.Collection;
import java.util.Collections;
public class CycleWindowAssigner extends WindowAssigner<CoGroupedStreams.TaggedUnion<Cycle, TravelEvent>, TimeWindow> {
@Override
public Collection<TimeWindow> assignWindows(CoGroupedStreams.TaggedUnion<Cycle, TravelEvent> element, long timestamp, WindowAssignerContext context) {
long start = DateUtils.dateToStamp(element.getOne().getBeginDateTime());
long end = DateUtils.dateToStamp(element.getOne().getEndDateTime());
return Collections.singletonList(new TimeWindow(start, end));
}
@Override
public Trigger<CoGroupedStreams.TaggedUnion<Cycle, TravelEvent>, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
return new CycleTimeTrigger();
}
@Override
public TypeSerializer<TimeWindow> getWindowSerializer(ExecutionConfig executionConfig) {
return new TimeWindow.Serializer();
}
@Override
public boolean isEventTime() {
return true;
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment