Skip to content

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

调整项目目录结构，提交一些测试代码，供参考。

parent 6e52c4b2
Showing with 1310 additions and 11 deletions
README.md 0 → 100644
### 项目简介
智能路网数据分析项目，各业务指标，如智能路网等业务数据均在此项目维护
### 项目结构
+ ZHHT-IRN-BD-ANALYSIS
+ offline 离线分析,主要基于Spark编程
+ holographic-intersection 全息路口项目
+ realtime 实时流处理，基于Flink编程
\ No newline at end of file
holographic-intersection/src/main/java/TestJava.java deleted 100644 → 0
public class TestJava {
public static void main(String[] args) {
System.out.println("test java");
}
}
holographic-intersection/pom.xml → offline/holographic-intersection/pom.xml
......@@ -3,7 +3,7 @@
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">
<parent>
<artifactId>ZHHT-IRN-BD-ANALYSIS</artifactId>
<artifactId>offline</artifactId>
<groupId>com.zhht.irn</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
......
offline/holographic-intersection/src/main/scala/com/zhht/irn/ExampleMetric1Job.scala 0 → 100644
package com.zhht.irn
import com.zhht.irn.config.ApplicationConfig.props
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import java.text.SimpleDateFormat
import java.util.{Date, Properties}
object ExampleMetric1Job {
// 读取信号数据
def get_sign(spark:SparkSession,TableName:String):DataFrame ={
val prop = new Properties()
prop.setProperty("user", props.getString("commons.datasource.mysql.username"))
prop.setProperty("password", props.getString("commons.datasource.mysql.password"))
prop.setProperty("driver", props.getString("commons.datasource.mysql.driver"))
val url = props.getString("commons.datasource.mysql.url")
spark.read.jdbc(url,TableName,prop)
}
// main class
def main(args: Array[String]): Unit = {
val sparkConfig = new SparkConf().setAppName("base_table")
sparkConfig.setMaster("local[*]") //在集群提交运行时，需要修改此行
val spark: SparkSession = SparkSession.builder().config(sparkConfig)
.config("metastore.catalog.default","hive")
.config("hive.strict.managed.tables","false")
.config("hive.create.as.insert.only","false")
.config("metastore.create.as.acid","false")
.config("spark.sql.warehouse.dir", "hdfs://localhost:8020/warehouse/tablespace/managed/hive")
.enableHiveSupport()
.getOrCreate()
val sc = spark.sparkContext
sc.setLogLevel("ERROR")
def NowTime():String = {
val now:Date = new Date()
val fm:SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
fm.format(now)
}
// project begin
println("Start Time:"+NowTime())
get_sign(spark,"dic_phase").show()
println("Finished Time:"+NowTime())
spark.stop()
sc.stop()
System.exit(0)
}
}
holographic-intersection/src/main/scala/com/zhht/irn/bigdata/atomic/base_tables.scala → offline/holographic-intersection/src/main/scala/com/zhht/irn/atomic/base_tables.scala
package com.zhht.irn.bigdata.atomic
package com.zhht.irn.atomic
import com.zhht.irn.bigdata.config.ApplicationConfig.props
import com.zhht.irn.config.ApplicationConfig.props
import java.util.{Date, Properties}
import org.apache.spark.SparkConf
......
holographic-intersection/src/main/scala/com/zhht/irn/bigdata/config/ApplicationConfig.scala → offline/holographic-intersection/src/main/scala/com/zhht/irn/config/ApplicationConfig.scala
package com.zhht.irn.bigdata.config
package com.zhht.irn.config
import com.typesafe.config.ConfigFactory
object ApplicationConfig {
val props = ConfigFactory.load("application.conf") //读取配置文件，获取配置信息
......
offline/pom.xml 0 → 100644
<?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">
<parent>
<artifactId>ZHHT-IRN-BD-ANALYSIS</artifactId>
<groupId>com.zhht.irn</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
<artifactId>offline</artifactId>
<!-- 此处用业务模块的全英文拼写代表相应的模块 -->
<modules>
<module>holographic-intersection</module>
</modules>
<properties>
<maven.compiler.source.version>1.8</maven.compiler.source.version>
<maven.compiler.target.version>1.8</maven.compiler.target.version>
<java.version>1.8</java.version>
<spark.version>2.4.0</spark.version>
<scala.version>2.11</scala.version>
<scala.binary.version>2.11.0</scala.binary.version>
</properties>
<dependencies>
<dependency>
<groupId>com.typesafe</groupId>
<artifactId>config</artifactId>
<version>1.2.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_${scala.version}</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_${scala.version}</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-mllib_${scala.version}</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_${scala.version}</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.27</version>
</dependency>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>${scala.binary.version}</version>
<scope>compile</scope>
</dependency>
<!-- <dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang4</artifactId>
<version>4.0</version>
</dependency>-->
<!--<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.32</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
<version>2.7</version>
</dependency>-->
<!--<dependency>
<groupId>com.fasterxml.woodstox</groupId>
<artifactId>woodstox-core</artifactId>
<version>5.0.3</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>3.1.0</version>
</dependency>-->
<!-- <dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_2.11</artifactId>
<version>1.12.0</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>1.12.0</version>
</dependency>
&lt;!&ndash; https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala &ndash;&gt;
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId>
<version>1.12.0</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.11</artifactId>
<version>1.12.0</version>
</dependency>
<dependency>
<groupId>org.roaringbitmap</groupId>
<artifactId>RoaringBitmap</artifactId>
<version>0.9.0</version>
</dependency>-->
</dependencies>
</project>
\ No newline at end of file
pom.xml
......@@ -8,8 +8,11 @@
<artifactId>ZHHT-IRN-BD-ANALYSIS</artifactId>
<packaging>pom</packaging>
<version>1.0-SNAPSHOT</version>
<!-- 智能路网相关数据分析代码均在此项目管理，第一层级以业务实时性来区分 -->
<modules>
<module>holographic-intersection</module>
<module>offline</module>
<module>realtime</module>
</modules>
<properties>
......
realtime/holographic-intersection-realtime/pom.xml 0 → 100644
<?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">
<parent>
<artifactId>realtime</artifactId>
<groupId>com.zhht.irn</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.zhht.irn</groupId>
<artifactId>holographic-intersection-realtime</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.47</version>
</dependency>
<dependency>
<groupId>org.apache.bahir</groupId>
<artifactId>flink-connector-redis_2.11</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
</dependencies>
</project>
\ No newline at end of file
realtime/holographic-intersection-realtime/src/main/java/com/zhht/irn/sink/PKMySQLSink.java 0 → 100644
package com.zhht.irn.sink;
import com.imooc.flink.utils.MySQLUtils;
import org.apache.flink.api.java.tuple.Tuple2;
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 PKMySQLSink extends RichSinkFunction<Tuple2<String, Double>> {
Connection connection;
PreparedStatement insertPstmt;
PreparedStatement updatePstmt;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
connection = MySQLUtils.getConnection();
insertPstmt = connection.prepareStatement("insert into pk_traffic(domain,traffic) values (?,?)");
updatePstmt = connection.prepareStatement("update pk_traffic set traffic=? where domain=?");
}
@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/holographic-intersection-realtime/src/main/java/com/zhht/irn/sink/PKRedisSink.java 0 → 100644
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/holographic-intersection-realtime/src/main/java/com/zhht/irn/sink/SinkApp.java 0 → 100644
package com.zhht.irn.sink;
import com.imooc.flink.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/holographic-intersection-realtime/src/main/java/com/zhht/irn/source/AccessSource.java 0 → 100644
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/holographic-intersection-realtime/src/main/java/com/zhht/irn/source/AccessSourceV2.java 0 → 100644
package com.zhht.irn.source;
import com.imooc.flink.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/holographic-intersection-realtime/src/main/java/com/zhht/irn/source/SourceApp.java 0 → 100644
package com.zhht.irn.source;
import com.imooc.flink.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/holographic-intersection-realtime/src/main/java/com/zhht/irn/source/Student.java 0 → 100644
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/holographic-intersection-realtime/src/main/java/com/zhht/irn/source/StudentSource.java 0 → 100644
package com.zhht.irn.source;
import com.imooc.flink.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();
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/holographic-intersection-realtime/src/main/java/com/zhht/irn/transformation/Access.java 0 → 100644
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/holographic-intersection-realtime/src/main/java/com/zhht/irn/transformation/PKMapFunction.java 0 → 100644
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/holographic-intersection-realtime/src/main/java/com/zhht/irn/transformation/TransformationApp.java 0 → 100644
package com.zhht.irn.transformation;
import com.imooc.flink.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/holographic-intersection-realtime/src/main/resources/application.conf 0 → 100644
##应用程序基础配置
application{
spark:{
##spark应用提交方式，如果未设，将接受spark_submit提交的参数
master:"local[*]"
// master:"yarn"
}
}
commons{
datasource:{
mysql:{
url:"jdbc:mysql://localhost:3307/test?useSSL=false&autoReconnect=true&failOverReadOnly=false&rewriteBatchedStatements=true&useUnicode=true&characterEncoding=utf8"
username:"root"
password:"mima"
driver:"com.mysql.cj.jdbc.Driver"
}
}
}
realtime/holographic-intersection-realtime/src/main/scala/com/zhht/irn/StreamingJob.scala 0 → 100644
///*
// * Licensed to the Apache Software Foundation (ASF) under one
// * or more contributor license agreements. See the NOTICE file
// * distributed with this work for additional information
// * regarding copyright ownership. The ASF licenses this file
// * to you under the Apache License, Version 2.0 (the
// * "License"); you may not use this file except in compliance
// * with the License. You may obtain a copy of the License at
// *
// * http://www.apache.org/licenses/LICENSE-2.0
// *
// * Unless required by applicable law or agreed to in writing, software
// * distributed under the License is distributed on an "AS IS" BASIS,
// * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// * See the License for the specific language governing permissions and
// * limitations under the License.
// */
//
//package com.zhht.irn
//
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
//
///**
// * Skeleton for a Flink Streaming Job.
// *
// * For a tutorial how to write a Flink streaming application, check the
// * tutorials and examples on the <a href="http://flink.apache.org/docs/stable/">Flink Website</a>.
// *
// * To package your application into a JAR file for execution, run
// * 'mvn clean package' on the command line.
// *
// * If you change the name of the main class (with the public static void main(String[] args))
// * method, change the respective entry in the POM.xml file (simply search for 'mainClass').
// */
//object StreamingJob {
// def main(args: Array[String]) {
//
// /**
// * 旅行信息
// *
// * 根据前端设备准实时的上报车辆的旅行信息
// * 实时提供车辆的旅行过程中的指标，目前尚未确定是实时提供旅行过程中的时刻信息，还是待车辆在路口的旅行过程完成后
// * 统一上报信息。
// *
// * 两种上报方式对应的计算逻辑也不相同：
// * 1、若为完成时统一上报，则只用采取来一条计算一条的逻辑，简单明确，但是会造成指标不够实时的问题。
// * 2、若为实时上报（可能是每200毫秒发一次车辆的状态信息），则需要实时计算，计算逻辑会稍微复杂一点，优势是准实时。
// *
// * 代码逻辑验证阶段，先从MySQL读取数据，按统一上报的方式来做
// */
//
// // 实时任务编程范式
//
// // 第一步：设置实时运行环境
// // settings
// val env = StreamExecutionEnvironment.getExecutionEnvironment
// val tableEnv = StreamTableEnvironment.create(env)
//
//// val name = "测试库"
//// val defaultDatabase = "test"
//// val username = "root"
//// val password = "mima"
//// val baseUrl = "jdbc:mysql//127.0.0.1:13305"
//
//
////
//// // 第二步：获取数据，目前基于Mysql做逻辑验证，后续改为Kafka
//// // connector接口参考：https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/connectors
////
//// val catalog = new JdbcCatalog(name, defaultDatabase, username, password, baseUrl)
//// tableEnv.registerCatalog("测试数据库", catalog)
////
//// tableEnv.useCatalog("测试数据库")
// val tableResult = tableEnv.executeSql("Create table cross_car_source (" +
// "ID bigint," +
// "CrossID string," +
// "CarID int," +
// "Locations string" +
// ")" +
// "WITH " +
// "('connector' = 'jdbc'," +
// "'url' = 'jdbc:mysql://127.0.0.1:13305/test'," +
// "'table-name' = 'tb_cross_cars_13070200137_20221025'," +
// "'username' = 'root'," +
// "'password' = 'mima')")
//
// // 第三步：进行数据处理，按照要求，计算出相应的指标
// // 结果要求是，根据当前旅行轨迹，计算该条轨迹在
// // tableEnv.from("tb_cross_cars_13070200137_20221025")
// val resultTable = tableEnv.sqlQuery("select * from cross_car_source")
// val resultStream: DataStream[CarTravelRecord] = tableEnv.toDataStream(resultTable, classOf[CarTravelRecord])
//
// // 以DataStream的API来处理了
//
//// tEnv
//// tableEnv.
// /*
// * Here, you can start creating your execution plan for Flink.
// *
// * Start with getting some data from the environment, like
// * env.readTextFile(textPath);
// *
// * then, transform the resulting DataStream[String] using operations
// * like
// * .filter()
// * .flatMap()
// * .join()
// * .group()
// *
// * and many more.
// * Have a look at the programming guide:
// *
// * http://flink.apache.org/docs/latest/apis/streaming/index.html
// *
// */
// }
//}
realtime/pom.xml 0 → 100644
<?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">
<parent>
<artifactId>ZHHT-IRN-BD-ANALYSIS</artifactId>
<groupId>com.zhht.irn</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
<artifactId>realtime</artifactId>
<modules>
<module>holographic-intersection-realtime</module>
</modules>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.17.1</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<version>2.17.1</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
<version>2.17.1</version>
</dependency>
</dependencies>
</dependencyManagement>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<flink.version>1.12.1</flink.version>
<scala.binary.version>2.12</scala.binary.version>
<target.java.version>1.8</target.java.version>
<maven.compiler.source>${target.java.version}</maven.compiler.source>
<maven.compiler.target>${target.java.version}</maven.compiler.target>
<log4j.version>2.12.1</log4j.version>
</properties>
</project>
\ No newline at end of file
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