第一章.Flink流处理核心编程

1.transform

一.简单滚动聚合算子

常见的滚动聚合算子 sum, min, max,minBy,maxBy

作用:KeyedStream的每一个支流做聚合,执行完成后,会将聚合的结果合成一个流返回,所以结果都是DataStream

参数:如果流中存储的是POJO或者scala的样例类, 参数使用字段名. 如果流中存储的是元组, 参数就是位置(基于0…)

返回:KeyedStream -> SingleOutputStreamOperator

  1. package com.atguigu.flink.day03;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.java.functions.KeySelector;
  5. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import java.util.ArrayList;
  10. /**
  11.  * 简单滚动聚合算子
  12.  */
  14. public class $01_RollingAggregationFunction {
  15.     public static void main(String[] args) throws Exception {
  16.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  17.         env.setParallelism(1);
  18.         ArrayList<WaterSensor> waterSensors = new ArrayList<>();
  19.         waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
  20.         waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
  21.         waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
  22.         waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
  23.         waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
  25.         DataStreamSource<WaterSensor> waterSensorDS = env.fromCollection(waterSensors);
  26.         waterSensorDS
  27.                 .keyBy(new KeySelector<WaterSensor, String>() {
  28.                     @Override
  29.                     public String getKey(WaterSensor value) throws Exception {
  30.                         return value.getId();
  31.                     }
  32.                 })
  33.                 /**
  34.                  * maxBy和minBy可以指定当出现相同值的时候,其他字段是否取第一个. 
  35.                  * true表示取第一个, false表示取与最大值(最小值)同一行的.
  36.                  * 5.3.10reduce
  37.                  */
  38.                 //.maxBy("vc",Boolean.FALSE).print();
  39.         //.sum("vc").print();
  40.         .min("vc").print();
  41.         env.execute();
  43.     }
  44. }

二.reduce

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

为什么还要把中间值也保存下来,考虑流式数据的特点:没有终点,也就没有最终的概念了,任何一个中间的聚合结果都是值

注意:聚合后结果的类型,必须和原来流中元素的类型保持一致

  1. package com.atguigu.flink.day03;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.MapFunction;
  5. import org.apache.flink.api.common.functions.ReduceFunction;
  6. import org.apache.flink.api.java.functions.KeySelector;
  7. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  8. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11. import java.util.ArrayList;
  13. /**
  14.  * reduce
  15.  *  1.keyby之后调用
  16.  *  2.每个分组的第一条数据来的时候,不会调用reduce方法,直接传递给下游
  17.  *  3.value1是上一次的计算结果,value2是当前来的数据
  18.  */
  20. public class $02_ReduceFunction {
  21.     public static void main(String[] args) throws Exception {
  22.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  23.         env.setParallelism(1);
  24.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  25.                 .socketTextStream("hadoop102", 9999)
  26.                 .map(new MapFunction<String, WaterSensor>() {
  27.                     @Override
  28.                     public WaterSensor map(String value) throws Exception {
  29.                         String[] line = value.split(",");
  30.                         return new WaterSensor(
  31.                                 line[0],
  32.                                 Long.parseLong(line[1]),
  33.                                 Integer.parseInt(line[2])
  34.                         );
  35.                     }
  36.                 });
  37.         sensorDS
  38.                 .keyBy(sensor -> sensor.getId())
  39.                 .reduce(new ReduceFunction<WaterSensor>() {
  40.                     @Override
  41.                     public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
  42.                         return new WaterSensor(
  43.                                 value1.getId(),
  44.                                 System.currentTimeMillis(),
  45.                                 value1.getVc() + value2.getVc()
  46.                         );
  47.                     }
  48.                 })
  49.                 .print();
  51.         env.execute();
  53.     }
  54. }

三.process

process算子在flink算是一个比较底层的算子,很多类型的流上都可以调用,可以从流中获取更多的信息(不仅仅是数据本身)

  1. package com.atguigu.flink.day03;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.MapFunction;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  8. import org.apache.flink.util.Collector;
  10. /**
  11.  * process
  12.  */
  14. public class $03_ProcessFunction {
  15.     public static void main(String[] args) throws Exception {
  16.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  17.         env.setParallelism(1);
  18.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  19.                 .socketTextStream("hadoop102", 9999)
  20.                 .map(new MapFunction<String, WaterSensor>() {
  21.                     @Override
  22.                     public WaterSensor map(String value) throws Exception {
  23.                         String[] line = value.split(",");
  24.                         return new WaterSensor(
  25.                                 line[0],
  26.                                 Long.parseLong(line[1]),
  27.                                 Integer.parseInt(line[2])
  28.                         );
  29.                     }
  30.                 });
  31.         sensorDS
  32.                 .keyBy(sensor -> sensor.getId())
  33.                 .process(new KeyedProcessFunction<String, WaterSensor, String>() {
  34.                     @Override
  35.                     public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
  36.                         out.collect(ctx.getCurrentKey());
  37.                     }
  38.                 })
  39.                 .print();
  41.         env.execute();
  43.     }
  44. }

四.对流重新分区的几个算子

  • keyBy: 先按照key分组,按照key的双重hash来选择后面的分区
  • shuffle: 对流中的元素随机分区
  • rebalance : 对流中的元素平均分布到每个区,当处理倾斜数据的时候,进行性能优化
  • rescale: 同rebalance一样,也是平均循环的分布数据,但是要比rebalance更高效,因为rescale不需要通过网络,完全走的管道

2.sink

一.KafkaSink

  1. pom.xml
  1. <dependency>
  2.      <groupId>org.apache.flink</groupId>
  3.      <artifactId>flink-connector-kafka_2.12</artifactId>
  4.      <version>1.13.1</version>
  5. </dependency>
  1. 代码
  1. package com.atguigu.flink.day03;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.MapFunction;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
  8. import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
  9. import org.apache.kafka.clients.producer.ProducerRecord;
  10. import javax.annotation.Nullable;
  11. import java.nio.charset.StandardCharsets;
  12. import java.util.Properties;
  14. /**
  15.  * 将数据写出到kafka
  16.  * 生产者:
  17.  *  ack:
  18.  *  buffersize: 默认 32M
  19.  *  batchsize: 默认 16K
  20.  *  linger.ms: 默认 0ms
  21.  *
  22.  *  注意事项:
  23.  *      1.可以指定一些生产者参数:ack.缓冲区,批次
  24.  *      2.默认分区器 FlinkFixedPartitioner
  25.  *      3.可以指定一致性级别: 至少一次,精准一次
  26.  */
  28. public class $04_KafkaSink {
  29.     public static void main(String[] args) throws Exception {
  30.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  31.         env.setParallelism(1);
  32.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  33.                 .socketTextStream("hadoop162", 9999)
  34.                 .map(new MapFunction<String, WaterSensor>() {
  35.                     @Override
  36.                     public WaterSensor map(String value) throws Exception {
  37.                         String[] line = value.split(",");
  38.                         return new WaterSensor(
  39.                                 line[0],
  40.                                 Long.parseLong(line[1]),
  41.                                 Integer.parseInt(line[2])
  42.                         );
  43.                     }
  44.                 });
  45.         Properties prop = new Properties();
  46.         prop.put("bootstrap.servers","hadoop162:9092,hadoop163:9092,hadoop164:9092");
  47.         sensorDS
  48.                 .addSink(new FlinkKafkaProducer<WaterSensor>(
  49.                         "default",
  50.                         new KafkaSerializationSchema<WaterSensor>() {
  51.                             @Override
  52.                             public ProducerRecord<byte[], byte[]> serialize(WaterSensor waterSensor, @Nullable Long aLong) {
  53.                                 return new ProducerRecord<byte[], byte[]>("waterSensor",waterSensor.toString().getBytes(StandardCharsets.UTF_8));
  54.                             }
  55.                         },
  56.                         prop,
  57.                         FlinkKafkaProducer.Semantic.EXACTLY_ONCE
  58.                 ));
  60.         env.execute();
  62.     }
  63. }

二.Redis Sink

  1. pom.xml
  1. <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-redis -->
  2. <dependency>
  3.      <groupId>org.apache.flink</groupId>
  4.      <artifactId>flink-connector-redis_2.11</artifactId>
  5.      <version>1.1.5</version>
  6. </dependency>
  1. 代码
  1. package com.atguigu.flink.day03;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.MapFunction;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
  8. import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
  9. import org.apache.flink.streaming.connectors.redis.RedisSink;
  10. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
  11. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
  12. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
  13. import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
  14. import org.apache.kafka.clients.producer.ProducerRecord;
  16. import javax.annotation.Nullable;
  17. import java.nio.charset.StandardCharsets;
  18. import java.util.Properties;
  20. /**
  21.  * 将数据写入到redis
  22.  */
  24. public class $05_RedisSink {
  25.     public static void main(String[] args) throws Exception {
  26.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  27.         env.setParallelism(1);
  28.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  29.                 .socketTextStream("hadoop162", 9999)
  30.                 .map(new MapFunction<String, WaterSensor>() {
  31.                     @Override
  32.                     public WaterSensor map(String value) throws Exception {
  33.                         String[] line = value.split(",");
  34.                         return new WaterSensor(
  35.                                 line[0],
  36.                                 Long.parseLong(line[1]),
  37.                                 Integer.parseInt(line[2])
  38.                         );
  39.                     }
  40.                 });
  41.         FlinkJedisPoolConfig jedisPoolConfig = new FlinkJedisPoolConfig.Builder()
  42.                 .setHost("hadoop162")
  43.                 .setPort(6379)
  44.                 .build();
  45.         RedisSink<WaterSensor> redisSinkFunction = new RedisSink<>(
  46.                 jedisPoolConfig,
  47.                 new RedisMapper<WaterSensor>() {
  48.                     @Override
  49.                     public RedisCommandDescription getCommandDescription() {
  50.                         //指定命令:往Hash结构插入数据(也可以是其他命令)
  51.                         return new RedisCommandDescription(RedisCommand.HSET, "waterSensor");
  52.                     }
  54.                     /**
  55.                      * 从数据读取hash结构的key
  56.                      * @param waterSensor
  57.                      * @return
  58.                      */
  59.                     @Override
  60.                     public String getKeyFromData(WaterSensor waterSensor) {
  61.                         return waterSensor.getId();
  62.                     }
  64.                     /**
  65.                      * 从数据提取,hash结构的value
  66.                      * @param waterSensor
  67.                      * @return
  68.                      */
  69.                     @Override
  70.                     public String getValueFromData(WaterSensor waterSensor) {
  71.                         return waterSensor.getVc().toString();
  72.                     }
  73.                 }
  74.         );
  75.         sensorDS.addSink(redisSinkFunction);
  77.         env.execute();
  79.     }
  80. }

三.ElasticsearchSink

  1. pom.xml
  1. <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-elasticsearch6 -->
  2. <dependency>
  3.     <groupId>org.apache.flink</groupId>
  4.     <artifactId>flink-connector-elasticsearch6_2.12</artifactId>
  5.     <version>1.13.1</version>
  6. </dependency>
  1. 代码
  1. package com.atguigu.flink.day03;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.MapFunction;
  5. import org.apache.flink.api.common.functions.RuntimeContext;
  6. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
  9. import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
  10. import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
  11. import org.apache.http.HttpHost;
  12. import org.elasticsearch.action.index.IndexRequest;
  13. import org.elasticsearch.client.Requests;
  15. import java.util.ArrayList;
  16. import java.util.HashMap;
  18. /**
  19.  * 将数据写入到es
  20.  * es: http://hadoop162:9200/watersensor/_search
  21.  * kibana: GET /watersensor/_search
  22.  */
  24. public class $06_ElasticsearchSink {
  25.     public static void main(String[] args) throws Exception {
  26.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  27.         env.setParallelism(1);
  28.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  29.                 .socketTextStream("hadoop162", 9999)
  30.                 .map(new MapFunction<String, WaterSensor>() {
  31.                     @Override
  32.                     public WaterSensor map(String value) throws Exception {
  33.                         String[] line = value.split(",");
  34.                         return new WaterSensor(
  35.                                 line[0],
  36.                                 Long.parseLong(line[1]),
  37.                                 Integer.parseInt(line[2])
  38.                         );
  39.                     }
  40.                 });
  41.         ArrayList<HttpHost> httpHosts = new ArrayList<>();
  42.         httpHosts.add(new HttpHost("hadoop162",9200));
  43.         httpHosts.add(new HttpHost("hadoop163",9200));
  44.         httpHosts.add(new HttpHost("hadoop164",9200));
  45.         ElasticsearchSink.Builder<WaterSensor> esBuilder = new ElasticsearchSink.Builder<>(
  46.                 httpHosts,
  47.                 new ElasticsearchSinkFunction<WaterSensor>() {
  48.                     @Override
  49.                     public void process(WaterSensor waterSensor, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
  50.                         HashMap<String, String> dataMap = new HashMap<>();
  51.                         dataMap.put("data", waterSensor.toString());
  52.                         IndexRequest indexRequest = Requests.indexRequest()
  53.                                 .index("watersensor")
  54.                                 .type("_doc")
  55.                                 .source(dataMap);
  56.                         requestIndexer.add(indexRequest);
  57.                     }
  58.                 }
  59.         );
  61.         //TODO 注意:这个仅仅是为了演示效果才设为1,生产环境不要改成1
  62.         esBuilder.setBulkFlushMaxActions(1);
  63.         ElasticsearchSink<WaterSensor> esSinkFunction = esBuilder.build();
  66.         sensorDS.addSink(esSinkFunction);
  67.         env.execute();
  69.     }
  70. }

day03[Flink流处理核心编程实战] - 图1

四.自定义sink

需求: 使用flink读取socket数据到mysql

  1. 导入mysql依赖
  1. <dependency>
  2.      <groupId>mysql</groupId>
  3.      <artifactId>mysql-connector-java</artifactId>
  4.      <version>5.1.49</version>
  5. </dependency>
  1. 建库建表
  1. create database test;
  2. use test;
  3. CREATE TABLE `sensor` (
  4.   `id` varchar(20) NOT NULL,
  5.   `ts` bigint(20) NOT NULL,
  6.   `vc` int(11) NOT NULL,
  7.   PRIMARY KEY (`id`,`ts`)
  8. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
  1. 代码
  1. package com.atguigu.flink.day03;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.MapFunction;
  5. import org.apache.flink.api.common.functions.RuntimeContext;
  6. import org.apache.flink.configuration.Configuration;
  7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
  10. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
  11. import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
  12. import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
  13. import org.apache.http.HttpHost;
  14. import org.elasticsearch.action.index.IndexRequest;
  15. import org.elasticsearch.client.Requests;
  17. import java.sql.Connection;
  18. import java.sql.DriverManager;
  19. import java.sql.PreparedStatement;
  20. import java.util.ArrayList;
  21. import java.util.HashMap;
  23. /**
  24.  * 自定义sink :将数据写出到kafka
  25.  * 生产环境用官方的JDBC写法
  26.  * https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/jdbc/
  27.  */
  29. public class $07_MySqlSink {
  30.     public static void main(String[] args) throws Exception {
  31.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  32.         env.setParallelism(1);
  33.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  34.                 .socketTextStream("hadoop162", 9999)
  35.                 .map(new MapFunction<String, WaterSensor>() {
  36.                     @Override
  37.                     public WaterSensor map(String value) throws Exception {
  38.                         String[] line = value.split(",");
  39.                         return new WaterSensor(
  40.                                 line[0],
  41.                                 Long.parseLong(line[1]),
  42.                                 Integer.parseInt(line[2])
  43.                         );
  44.                     }
  45.                 });
  48.         sensorDS.addSink(new MySink());
  49.         env.execute();
  51.     }
  52.     public static class MySink extends RichSinkFunction<WaterSensor>{
  54.         private Connection connection;
  55.         private PreparedStatement preparedStatement;
  57.         @Override
  58.         public void open(Configuration parameters) throws Exception {
  59.             connection = DriverManager.getConnection("jdbc:mysql://hadoop162:3306/test", "root", "aaaaaa");
  60.             preparedStatement = connection.prepareStatement("insert into sensor values (?,?,?)");
  61.         }
  63.         @Override
  64.         public void close() throws Exception {
  65.             if(preparedStatement != null){
  66.                 preparedStatement.close();
  67.             }
  68.             if(connection != null){
  69.                 connection.close();
  70.             }
  71.         }
  73.         @Override
  74.         public void invoke(WaterSensor value, Context context) throws Exception {
  75.             preparedStatement.setString(1, value.getId());
  76.             preparedStatement.setLong(2,value.getTs());
  77.             preparedStatement.setInt(3,value.getVc());
  78.             preparedStatement.execute();
  79.         }
  80.     }
  81. }

3.配置BATH执行模式

执行模式有3个选择可配

  1. STREAMING(默认)
  2. BATCH
  3. AUTOMATIC
  1. 有界数据用哪个STREAMING 和BATCH的区别
  2.     STREAMING模式下, 数据是来一条输出一次结果.
  3.     BATCH模式下, 数据处理完之后, 一次性输出结果.

命令行配置

bin/flink run -Dexecution.runtime-mode=BATCH …

第二章.Flink流处理核心编程实战

1.基于埋点日志数据的网络流量统计

一.网站总浏览量(PV)的统计

  1.     衡量网站流量一个最简单的指标，就是网站的页面浏览量（Page View，PV）。用户每次打开一个页面便记录1次PV，多次打开同一页面则浏览量累计。
  2.     一般来说，PV与来访者的数量成正比，但是PV并不直接决定页面的真实来访者数量，如同一个来访者通过不断的刷新页面，也可以制造出非常高的PV。接下来我们就用咱们之前学习的Flink算子来实现PV的统计
  1. 用于封装数据的javabean
  1. package com.atguigu.flink.day03.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. @Data
  8. @NoArgsConstructor
  9. @AllArgsConstructor
  10. public class UserBehavior {
  11.     /**
  12.      * 用户ID
  13.      */
  14.     private Long userId;
  16.     /**
  17.      * 商品ID
  18.      */
  19.     private Long itemId;
  21.     /**
  22.      * 品类ID
  23.      */
  24.     private Integer categoryId;
  26.     /**
  27.      * 用户的行为：pv、buy、cart、fav
  28.      */
  29.     private String behavior;
  31.     /**
  32.      * 时间戳
  33.      */
  34.     private Long timestamp;
  35. }
  1. 代码
  1. package com.atguigu.flink.day03;
  4. import com.atguigu.flink.day03.pojo.UserBehavior;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.streaming.api.functions.ProcessFunction;
  10. import org.apache.flink.util.Collector;
  14. /**
  15.  * pv
  16.  */
  18. public class $08_PV {
  19.     public static void main(String[] args) throws Exception {
  20.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  21.         env.setParallelism(2);
  22.         //1.读取数据
  23.         SingleOutputStreamOperator<UserBehavior> userBehavioirDS = env
  24.                 .readTextFile("input/UserBehavior.csv")
  25.                 .map(new MapFunction<String, UserBehavior>() {
  26.                     @Override
  27.                     public UserBehavior map(String value) throws Exception {
  28.                         String[] datas = value.split(",");
  29.                         return new UserBehavior(
  30.                                 Long.parseLong(datas[0]),
  31.                                 Long.parseLong(datas[1]),
  32.                                 Integer.parseInt(datas[2]),
  33.                                 datas[3],
  34.                                 Long.parseLong(datas[4])
  35.                         );
  36.                     }
  37.                 });
  38.         //2.处理数据
  39.         userBehavioirDS
  40.                 .filter(user -> "pv".equals(user.getBehavior()))
  41.                 .process(new ProcessFunction<UserBehavior, Integer>() {
  42.                     int pvCount = 0;
  43.                     @Override
  44.                     public void processElement(UserBehavior value, Context ctx, Collector<Integer> out) throws Exception {
  45.                         pvCount++;
  46.                         out.collect(pvCount);
  47.                     }
  48.                 }).setParallelism(1)
  49.                 .print();
  51.         env.execute();
  53.     }
  55.     }

二.网站独立访客数(UV)的统计

上一个案例中，我们统计的是所有用户对页面的所有浏览行为，也就是说，同一用户的浏览行为会被重复统计。而在实际应用中，我们往往还会关注，到底有多少不同的用户访问了网站，所以另外一个统计流量的重要指标是网站的独立访客数（Unique Visitor，UV）

对于UserBehavior数据源来说，我们直接可以根据userId来区分不同的用户.

代码

  1. package com.atguigu.flink.day03;
  4. import com.atguigu.flink.day03.pojo.UserBehavior;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.functions.ProcessFunction;
  10. import org.apache.flink.util.Collector;
  11. import java.util.HashSet;
  12. import java.util.Set;
  15. /**
  16.  * uv
  17.  */
  19. public class $09_UV {
  20.     public static void main(String[] args) throws Exception {
  21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  22.         env.setParallelism(2);
  23.         //1.读取数据
  24.         SingleOutputStreamOperator<UserBehavior> userBehavioirDS = env
  25.                 .readTextFile("input/UserBehavior.csv")
  26.                 .map(new MapFunction<String, UserBehavior>() {
  27.                     @Override
  28.                     public UserBehavior map(String value) throws Exception {
  29.                         String[] datas = value.split(",");
  30.                         return new UserBehavior(
  31.                                 Long.parseLong(datas[0]),
  32.                                 Long.parseLong(datas[1]),
  33.                                 Integer.parseInt(datas[2]),
  34.                                 datas[3],
  35.                                 Long.parseLong(datas[4])
  36.                         );
  37.                     }
  38.                 });
  40.         //2.处理数据,提取出 用户ID,存在SET结构里,求SET的长度就是UV值
  41.         userBehavioirDS
  42.                 .filter(user -> "pv".equals(user.getBehavior()))
  43.                 .process(new ProcessFunction<UserBehavior, Integer>() {
  44.                     Set<Long> uvSet = new HashSet<>();
  45.                     @Override
  46.                     public void processElement(UserBehavior value, Context ctx, Collector<Integer> out) throws Exception {
  47.                        //每来一条数据,提取出用户ID,存到Set里
  48.                         uvSet.add(value.getItemId());
  49.                         //将Set的元素个数(uv值),传递给下游
  50.                         out.collect(uvSet.size());
  51.                     }
  52.                 }).setParallelism(1)
  53.                 .print();
  55.         env.execute();
  57.     }
  59.     }

2.市场营销商业指标统计分析

随着智能手机的普及，在如今的电商网站中已经有越来越多的用户来自移动端，相比起传统浏览器的登录方式，手机APP成为了更多用户访问电商网站的首选。对于电商企业来说，一般会通过各种不同的渠道对自己的APP进行市场推广，而这些渠道的统计数据（比如，不同网站上广告链接的点击量、APP下载量）就成了市场营销的重要商业指标。

一.APP市场推广统计 -分渠道

  1. 封装数据的javabean
  1. package com.atguigu.flink.day03.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. @Data
  8. @AllArgsConstructor
  9. @NoArgsConstructor
  10. public class MarketingUserBehavior {
  11.     private Long userId;
  12.     private String behavior;
  13.     private String channel;
  14.     private Long timestamp;
  15. }
  1. 具体实现代码
  1. package com.atguigu.flink.day03;
  4. import com.atguigu.flink.day03.pojo.MarketingUserBehavior;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.api.java.tuple.Tuple2;
  7. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.functions.source.SourceFunction;
  10. import java.util.Arrays;
  11. import java.util.List;
  12. import java.util.Random;
  16. /**
  17.  * APP市场推广统计-分渠道
  18.  */
  20. public class $10_AppAnalysisByChannel {
  21.     public static void main(String[] args) throws Exception {
  22.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  23.         env.setParallelism(2);
  24.         //1.读取数据
  25.         DataStreamSource<MarketingUserBehavior> inputDS = env.addSource(new AppSource());
  26.         //2.处理数据:按照不同渠道对不同行为进行统计-->对两个维度进行keyby
  27.         inputDS
  28.                 .map(new MapFunction<MarketingUserBehavior, Tuple2<String,Integer>>() {
  29.                     @Override
  30.                     public Tuple2<String, Integer> map(MarketingUserBehavior value) throws Exception {
  31.                         return Tuple2.of(value.getChannel() +"_"+value.getBehavior(),1);
  32.                     }
  33.                 })
  34.                 .keyBy(t->t.f0)
  35.                 .sum(1)
  36.                 .print();
  39.         env.execute();
  41.     }
  42.     public static class AppSource implements SourceFunction<MarketingUserBehavior>{
  43.         static boolean flag = true;
  44.         List<String> behaviorList = Arrays.asList("DOWNLOAD", "INSTALL", "UPDATE", "UNINSTALL");
  45.         List<String> channelList = Arrays.asList("XIAOMI", "HUAWEI", "OPPO", "VIVO");
  48.         @Override
  49.         public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
  50.             Random random = new Random();
  51.             while (flag) {
  52.                 ctx.collect(
  53.                         new MarketingUserBehavior(
  54.                                 random.nextLong(),
  55.                                 behaviorList.get(random.nextInt(behaviorList.size())),
  56.                                 channelList.get(random.nextInt(channelList.size())),
  57.                                 System.currentTimeMillis()
  59.                         )
  60.                 );
  61.                 Thread.sleep(1000);
  62.             }
  63.         }
  65.         @Override
  66.         public void cancel() {
  67.            flag = false;
  68.         }
  69.     }
  71.     }

二.APP市场推广统计 -不分渠道

  1. 实现代码
  1. package com.atguigu.flink.day03;
  4. import com.atguigu.flink.day03.pojo.MarketingUserBehavior;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.api.java.tuple.Tuple2;
  7. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.functions.source.SourceFunction;
  11. import java.util.Arrays;
  12. import java.util.List;
  13. import java.util.Random;
  16. /**
  17.  * APP市场推广统计-不分渠道
  18.  */
  20. public class $11_AppAnalysisWithOutChannel {
  21.     public static void main(String[] args) throws Exception {
  22.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  23.         env.setParallelism(2);
  24.         //1.读取数据
  25.         DataStreamSource<MarketingUserBehavior> inputDS = env.addSource(new AppSource());
  26.         //2.处理数据:按照不同渠道对不同行为进行统计-->对两个维度进行keyby
  27.         inputDS
  28.                 .map(new MapFunction<MarketingUserBehavior, Tuple2<String,Integer>>() {
  29.                     @Override
  30.                     public Tuple2<String, Integer> map(MarketingUserBehavior value) throws Exception {
  31.                         return Tuple2.of(value.getBehavior(),1);
  32.                     }
  33.                 })
  34.                 .keyBy(t->t.f0)
  35.                 .sum(1)
  36.                 .print();
  39.         env.execute();
  41.     }
  42.     public static class AppSource implements SourceFunction<MarketingUserBehavior>{
  43.         static boolean flag = true;
  44.         List<String> behaviorList = Arrays.asList("DOWNLOAD", "INSTALL", "UPDATE", "UNINSTALL");
  45.         List<String> channelList = Arrays.asList("XIAOMI", "HUAWEI", "OPPO", "VIVO");
  48.         @Override
  49.         public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
  50.             Random random = new Random();
  51.             while (flag) {
  52.                 ctx.collect(
  53.                         new MarketingUserBehavior(
  54.                                 random.nextLong(),
  55.                                 behaviorList.get(random.nextInt(behaviorList.size())),
  56.                                 channelList.get(random.nextInt(channelList.size())),
  57.                                 System.currentTimeMillis()
  59.                         )
  60.                 );
  61.                 Thread.sleep(1000);
  62.             }
  63.         }
  65.         @Override
  66.         public void cancel() {
  67.            flag = false;
  68.         }
  69.     }
  71.     }

3.各省市页面广告点击量实时统计

  1. 电商网站的市场营销商业指标中，除了自身的APP推广，还会考虑到页面上的广告投放（包括自己经营的产品和其它网站的广告）。所以广告相关的统计分析，也是市场营销的重要指标。<br />    对于广告的统计，最简单也最重要的就是页面广告的点击量，网站往往需要根据广告点击量来制定定价策略和调整推广方式，而且也可以借此收集用户的偏好信息。更加具体的应用是，我们可以根据用户的地理位置进行划分，从而总结出不同省份用户对不同广告的偏好，这样更有助于广告的精准投放。
  1. 用于封装数据的javabean
  1. package com.atguigu.flink.day03.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. @Data
  8. @AllArgsConstructor
  9. @NoArgsConstructor
  10. public class AdsClickLog {
  11.     private Long userId;
  12.     private Long adId;
  13.     private String province;
  14.     private String city;
  15.     private Long timestamp;
  16. }
  1. 实现代码
  1. package com.atguigu.flink.day03;
  4. import com.atguigu.flink.day03.pojo.AdsClickLog;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.api.java.functions.KeySelector;
  7. import org.apache.flink.api.java.tuple.Tuple2;
  8. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  11. import org.apache.flink.util.Collector;
  13. import java.util.*;
  16. /**
  17.  * 各省市页面广告点击量实时统计
  18.  *
  19.  * 扩展知识:
  20.  *      函数类中,定义的变量个数和作用范围
  21.  *      1.函数类创建了几次?
  22.  *          每个并行实例new一次
  23.  *      2.类中定义的变量一共有几个?
  24.  *          每个并行实例一份
  25.  *      3.函数类的方法调用几次?
  26.  *          大部分是,来一条数据调用一次方法
  27.  *          也有一些是,来一批数据调用一次(比如窗口的全量函数)
  28.  */
  30. public class $12_AdClick {
  31.     public static void main(String[] args) throws Exception {
  32.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  33.         env.setParallelism(2);
  34.         //1.读取数据,转换格式
  35.         SingleOutputStreamOperator<AdsClickLog> adClickDS = env
  36.                 .readTextFile("input/AdClickLog.csv")
  37.                 .map(new MapFunction<String, AdsClickLog>() {
  38.                     @Override
  39.                     public AdsClickLog map(String value) throws Exception {
  40.                         String[] datas = value.split(",");
  41.                         return new AdsClickLog(
  42.                                 Long.parseLong(datas[0]),
  43.                                 Long.parseLong(datas[1]),
  44.                                 datas[2],
  45.                                 datas[3],
  46.                                 Long.parseLong(datas[4])
  48.                         );
  49.                     }
  50.                 });
  52.         //2.处理数据
  53.         adClickDS
  54.                 .keyBy(new KeySelector<AdsClickLog, Tuple2<String,Long>>() {
  55.                     @Override
  56.                     public Tuple2<String, Long> getKey(AdsClickLog value) throws Exception {
  57.                         return Tuple2.of(value.getProvince(),value.getAdId());
  58.                     }
  59.                 })
  60.                 .process(new KeyedProcessFunction<Tuple2<String, Long>, AdsClickLog, String>() {
  61.                     Map<Tuple2<String, Long>, Integer> map = new HashMap<>();
  62.                     @Override
  63.                     public void processElement(AdsClickLog value, Context ctx, Collector<String> out) throws Exception {
  64.                         Integer count = map.get(ctx.getCurrentKey());
  65.                         if(count == null){
  66.                             //说明当前的key不存在,说明这是第一条数据
  67.                             count = 1;
  68.                         }else{
  69.                             //说明当前的key存在,说明这不是第一条数据,正常+1就行
  70.                             count++;
  71.                         }
  72.                         map.put(ctx.getCurrentKey(),count);
  73.                         out.collect(map.toString());
  74.                     }
  75.                 })
  76.                 .print();
  77.         env.execute();
  78.     }
  80.     }

4.订单支付实时监控

  1. 在电商网站中，订单的支付作为直接与营销收入挂钩的一环，在业务流程中非常重要。对于订单而言，为了正确控制业务流程，也为了增加用户的支付意愿，网站一般会设置一个支付失效时间，超过一段时间不支付的订单就会被取消。另外，对于订单的支付，我们还应保证用户支付的正确性，这可以通过第三方支付平台的交易数据来做一个实时对账。

需求:来自两条流的订单交易匹配

  1. 对于订单支付事件，用户支付完成其实并不算完，我们还得确认平台账户上是否到账了。而往往这会来自不同的日志信息，所以我们要同时读入两条流的数据来做合并处理。
  1. 用于封装数据的javabean
  1. package com.atguigu.flink.day03.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. @Data
  8. @AllArgsConstructor
  9. @NoArgsConstructor
  10. public class OrderEvent {
  11.     private Long orderId;
  12.     private String eventType;
  13.     private String txId;
  14.     private Long eventTime;
  15. }
  1. package com.atguigu.flink.day03.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. @Data
  8. @AllArgsConstructor
  9. @NoArgsConstructor
  10. public class TxEvent {
  11.     private String txId;
  12.     private String payChannel;
  13.     private Long eventTime;
  14. }
  1. 实现代码
  1. package com.atguigu.flink.day03;
  5. import com.atguigu.flink.day03.pojo.OrderEvent;
  6. import com.atguigu.flink.day03.pojo.TxEvent;
  7. import org.apache.flink.api.common.functions.MapFunction;
  8. import org.apache.flink.streaming.api.datastream.ConnectedStreams;
  9. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11. import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
  12. import org.apache.flink.util.Collector;
  14. import java.util.HashMap;
  15. import java.util.Map;
  18. /**
  19.  *connect连接两条流的时候,如果需要做一些比对操作,要考虑数据的分布,
  20.  * 需要比对的数据要在同一个分区里  ---->keyby(关联条件)
  21.  */
  23. public class $13_OrderTxDetect {
  24.     public static void main(String[] args) throws Exception {
  25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  26.         env.setParallelism(1);
  27.         //1.读取数据,转换格式
  28.         SingleOutputStreamOperator<OrderEvent> orderDS = env.readTextFile("input/OrderLog.csv")
  29.                 .map(new MapFunction<String, OrderEvent>() {
  30.                     @Override
  31.                     public OrderEvent map(String value) throws Exception {
  32.                         String[] datas = value.split(",");
  33.                         return new OrderEvent(
  34.                                 Long.parseLong(datas[0]),
  35.                                 datas[1],
  36.                                 datas[2],
  37.                                 Long.parseLong(datas[3])
  38.                         );
  39.                     }
  40.                 });
  42.         SingleOutputStreamOperator<TxEvent> txDS = env.readTextFile("input/ReceiptLog.csv")
  43.                 .map(new MapFunction<String, TxEvent>() {
  44.                     @Override
  45.                     public TxEvent map(String value) throws Exception {
  46.                         String[] datas = value.split(",");
  47.                         return new TxEvent(
  48.                                 datas[0],
  49.                                 datas[1],
  50.                                 Long.parseLong(datas[2])
  51.                         );
  52.                     }
  53.                 });
  56.         //2.数据处理,需要做一个合流操作
  57.         ConnectedStreams<OrderEvent, TxEvent> orderTxCS = orderDS.connect(txDS);
  58.         ConnectedStreams<OrderEvent, TxEvent> orderTxKCS = orderTxCS.keyBy(order -> order.getTxId(), tx -> tx.getTxId());
  60.         SingleOutputStreamOperator<String> resultDS = orderTxKCS.process(new CoProcessFunction<OrderEvent, TxEvent, String>() {
  61.             Map<String, OrderEvent> orderMap = new HashMap<>();
  62.             Map<String, TxEvent> txMap = new HashMap<>();
  64.             /**
  65.              * 处理的是业务系统的数据
  66.              * @param value
  67.              * @param ctx
  68.              * @param out
  69.              * @throws Exception
  70.              */
  71.             @Override
  72.             public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
  73.                 //进入这个方法,说明当前来的是业务系统的数据,先查一下,对方交易(数据)来了没有
  74.                 TxEvent txEvent = txMap.get(value.getTxId());
  75.                 if (txEvent == null) {
  76.                     //说明交易数据没来过,等他先把自己存起来
  77.                     orderMap.put(value.getTxId(), value);
  79.                 } else {
  80.                     //说明,对方来过==>匹配上,删除map中的数据
  81.                     out.collect("订单" + value.getOrderId() + "对账成功");
  82.                     txMap.remove(value.getTxId());
  84.                 }
  85.             }
  87.             /**
  88.              * 处理的是交易系统的数据
  89.              * @param value
  90.              * @param ctx
  91.              * @param out
  92.              * @throws Exception
  93.              */
  94.             @Override
  95.             public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
  96.                 //进入这个方法,说明当前来的是交易系统的数据,先查一下,对方业务(数据)来了没有
  97.                 OrderEvent orderEvent = orderMap.get(value.getTxId());
  98.                 if (orderEvent == null) {
  99.                     //说明业务数据没来过,等他先把自己存起来
  100.                     txMap.put(value.getTxId(), value);
  102.                 } else {
  103.                     //说明,对方来过==>匹配上,删除map中的数据
  104.                     out.collect("订单" + orderEvent.getOrderId() + "对账成功");
  105.                     orderMap.remove(value.getTxId());
  107.                 }
  108.             }
  109.         });
  110.         resultDS.print();
  113.         env.execute();
  114.     }
  116.     }