• java Flink（四十一）Flink+avro+广播流broadcast实现流量的清洗

    背景简介

    本文简单模拟对流量的处理，大概步骤如下：

    1、通过获取一个维度流，内容是流量内容的元数据信息，获取解析并进行广播

    2、获取实时流量流，做延迟处理（防止数据关联不上）

    3、流量流关联元数据广播流，通过元数据信息获取对应的数据

    4、打包成avro格式（自行百度）数据并进行sink

    代码开发

    1、pom引入

    flink版本1.14

    1. "1.0" encoding="UTF-8"?>
    2. <project xmlns="http://maven.apache.org/POM/4.0.0"
    3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    5.     <modelVersion>4.0.0modelVersion>
    6.  
    7.     <groupId>org.examplegroupId>
    8.     <artifactId>FlinkCodeartifactId>
    9.     <version>1.0-SNAPSHOTversion>
    10.  
    11.     <properties>
    12.         <maven.compiler.source>8maven.compiler.source>
    13.         <maven.compiler.target>8maven.compiler.target>
    14.         <jdk.version>1.8jdk.version>
    15.         <jar.name>ubs-data-converterjar.name>
    16.         <project.build.sourceEncoding>UTF-8project.build.sourceEncoding>
    17.         
    18.         <flink.version>1.14.4flink.version>
    19.     properties>
    20.     <dependencies>
    21.         <dependency>
    22.             <groupId>org.apache.flinkgroupId>
    23.             <artifactId>flink-streaming-java_2.11artifactId>
    24.             <version>${flink.version}version>
    25.             <scope>providedscope>
    26.         dependency>
    27.         <dependency>
    28.             <groupId>org.apache.flinkgroupId>
    29.             <artifactId>flink-connector-kafka_2.11artifactId>
    30.             <version>${flink.version}version>
    31.             <exclusions>
    32.                 <exclusion>
    33.                     <groupId>org.apache.kafkagroupId>
    34.                     <artifactId>kafka-clientsartifactId>
    35.                 exclusion>
    36.             exclusions>
    37.         dependency>
    38.         <dependency>
    39.             <groupId>org.apache.flinkgroupId>
    40.             <artifactId>flink-runtime-web_2.11artifactId>
    41.             <version>${flink.version}version>
    42.             <scope>providedscope>
    43.         dependency>
    44.         <dependency>
    45.             <groupId>org.apache.httpcomponentsgroupId>
    46.             <artifactId>httpclientartifactId>
    47.             <version>4.5.10version>
    48.         dependency>
    49.         <dependency>
    50.             <groupId>com.alibabagroupId>
    51.             <artifactId>fastjsonartifactId>
    52.             <version>1.2.8version>
    53.         dependency>
    54.         <dependency>
    55.             <groupId>org.apache.avrogroupId>
    56.             <artifactId>avroartifactId>
    57.             <version>1.9.2version>
    58.         dependency>
    59.         <dependency>
    60.             <groupId>org.apache.httpcomponentsgroupId>
    61.             <artifactId>httpcoreartifactId>
    62.             <version>4.4.1version>
    63.         dependency>
    64.         <dependency>
    65.             <groupId>org.projectlombokgroupId>
    66.             <artifactId>lombokartifactId>
    67.             <version>1.18.16version>
    68.         dependency>
    69.         <dependency>
    70.             <groupId>org.projectlombokgroupId>
    71.             <artifactId>lombokartifactId>
    72.             <version>1.18.16version>
    73.             <scope>compilescope>
    74.         dependency>
    75.     dependencies>
    76.  
    77.     <build>
    78.         <plugins>
    79.             <plugin>
    80.                 <groupId>org.apache.avrogroupId>
    81.                 <artifactId>avro-maven-pluginartifactId>
    82.                 <version>1.9.2version>
    83.                 <executions>
    84.                     <execution>
    85.                         <phase>generate-sourcesphase>
    86.                         <goals>
    87.                             <goal>schemagoal>
    88.                         goals>
    89.                         <configuration>
    90.                             <sourceDirectory>${project.basedir}/src/main/resources/sourceDirectory>
    91.                             <outputDirectory>${project.basedir}/src/main/java/com/msxfoutputDirectory>
    92.                         configuration>
    93.                     execution>
    94.                 executions>
    95.             plugin>
    96.             <plugin>
    97.                 <groupId>org.apache.maven.pluginsgroupId>
    98.                 <artifactId>maven-compiler-pluginartifactId>
    99.                 <version>3.1version>
    100.                 <configuration>
    101.                     <source>${jdk.version}source>
    102.                     <target>${jdk.version}target>
    103.                     <encoding>${project.build.sourceEncoding}encoding>
    104.                 configuration>
    105.             plugin>
    106.             <plugin>
    107.                 <groupId>org.apache.maven.pluginsgroupId>
    108.                 <artifactId>maven-shade-pluginartifactId>
    109.                 <version>3.1.1version>
    110.                 <executions>
    111.                     <execution>
    112.                         <phase>packagephase>
    113.                         <goals>
    114.                             <goal>shadegoal>
    115.                         goals>
    116.                         <configuration>
    117.                             <finalName>${jar.name}finalName>
    118.                             <artifactSet>
    119.                                 <excludes>
    120.                                     <exclude>com.google.code.findbugs:jsr305exclude>
    121.                                     <exclude>org.slf4j:*exclude>
    122.                                     <exclude>log4j:*exclude>
    123.                                     <exclude>org.glassfish.jersey.core:jersey-commonexclude>
    124.                                 excludes>
    125.                             artifactSet>
    126.                             <relocations>
    127.                                 <relocation>
    128.                                     <pattern>com.google.commonpattern>
    129.                                     <shadedPattern>com.shade.google.commonshadedPattern>
    130.                                 relocation>
    131.                                 <relocation>
    132.                                     <pattern>org.apache.kafkapattern>
    133.                                     <shadedPattern>org.shade.apache.kafkashadedPattern>
    134.                                 relocation>
    135.                             relocations>
    136.                             <filters>
    137.                                 <filter>
    138.                                     <artifact>*artifact>
    139.                                     <includes>
    140.                                         <include>org/apache/htrace/**include>
    141.                                         <include>org/apache/avro/**include>
    142.                                         <include>com/msxf/**include>
    143.                                         <include>org/apache/flink/streaming/**include>
    144.                                         <include>org/apache/flink/connector/**include>
    145.                                         <include>org/apache/kafka/**include>
    146.                                         <include>org/apache/hive/**include>
    147.                                         <include>org/apache/hadoop/hive/**include>
    148.                                         <include>org/apache/curator/**include>
    149.                                         <include>org/apache/zookeeper/**include>
    150.                                         <include>org/apache/jute/**include>
    151.                                         <include>org/apache/thrift/**include>
    152.                                         <include>org/apache/http/**include>
    153.                                         <include>org/I0Itec/**include>
    154.                                         <include>jline/**include>
    155.                                         <include>com/yammer/**include>
    156.                                         <include>kafka/**include>
    157.                                         <include>org/apache/hadoop/hbase/**include>
    158.                                         <include>com/alibaba/fastjson/**include>
    159.                                         <include>org/elasticsearch/action/**include>
    160.                                         <include>io/confluent/**include>
    161.                                         <include>com/fasterxml/**include>
    162.                                         <include>org/elasticsearch/**include>
    163.                                         <include>hbase-default.xmlinclude>
    164.                                         <include>hbase-site.xmlinclude>
    165.                                     includes>
    166.                                 filter>
    167.                                 <filter>
    168.                                     <artifact>org.apache.hadoop.hive.*:*artifact>
    169.                                     <excludes>
    170.                                         <exclude>exclude>
    171.                                         <exclude>exclude>
    172.                                         <exclude>exclude>
    173.                                     excludes>
    174.                                 filter>
    175.                             filters>
    176.                         configuration>
    177.                     execution>
    178.                 executions>
    179.             plugin>
    180.         plugins>
    181.     build>
    182. project>

    2、模拟元数据流

    正常数据流应该通过其他方式（比如访问数据库、KAFKA流）获取，本次我们直接自定义source

    1. package source;
    2.  
    3. import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
    4.  
    5.  
    6. public class StringSource extends RichSourceFunction {
    7.  
    8.     Boolean running = true;
    9.     @Override
    10.     public void run(SourceContext ctx) throws Exception {
    11.         while (running){
    12.             String value = String.format("{\"data\":[{\"name\":\"id\",\"comment\":\"ID\"}" +
    13.                     ",{\"name\":\"age\",\"comment\":\"年龄\"}" +
    14.                     ",{\"name\":\"sex\",\"comment\":\"性别\"}]}");
    15.             ctx.collect(value);
    16.             running=false;
    17.         }
    18.     }
    19.  
    20.     @Override
    21.     public void cancel() {
    22.         running=false;
    23.     }
    24. }

    3、模拟流量流

    模拟流量数据，本身也是通过KAFKA获取实时流量数据，本文是简单Demo，所以也通过自定义Source获取

    1. package source;
    2.  
    3. import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
    4.  
    5. import java.util.Arrays;
    6. import java.util.List;
    7. import java.util.Random;
    8.  
    9.  
    10. public class FlowSourceFunction extends RichSourceFunction {
    11.  
    12.     Boolean running=true;
    13.  
    14.     private final List USERS = Arrays.asList("张三","李四","牛二");
    15.  
    16.     private final List BEHAVIOR = Arrays.asList("login", "out", "delete");
    17.  
    18.     private final List SEX = Arrays.asList("男", "女");
    19.  
    20.     Random random = new Random();
    21.  
    22.     @Override
    23.     public void run(SourceContext ctx) throws Exception {
    24.         while (running){
    25.             String id = USERS.get(random.nextInt(USERS.size()));
    26.             String age = String.valueOf(random.nextInt(100));
    27.             String sex = SEX.get(random.nextInt(SEX.size()));
    28.             String time = String.valueOf(System.currentTimeMillis());
    29.             String res = String.format("{\"id\":\"%s\"," +
    30.                     "\"age\":\"%s\"," +
    31.                     "\"sex\":\"%s\"," +
    32.                     "\"time\":\"%s\"}",id,age,sex,time);
    33.             ctx.collect(res);
    34.             Thread.sleep(1000);
    35.         }
    36.     }
    37.  
    38.     @Override
    39.     public void cancel() {
    40.         running=false;
    41.     }
    42. }

    4、MapFunction处理元数据信息

    处理元数据信息

    1. package func;
    2.  
    3. import bean.SchemaInfo;
    4. import com.alibaba.fastjson.JSON;
    5. import com.alibaba.fastjson.JSONObject;
    6. import org.apache.avro.Schema;
    7. import org.apache.flink.api.common.functions.RichMapFunction;
    8. import org.apache.avro.SchemaBuilder;
    9. import org.apache.flink.util.StringUtils;
    10.  
    11. import java.util.HashSet;
    12. import java.util.stream.Stream;
    13.  
    14.  
    15. public class MetaDataMapFunction extends RichMapFunction {
    16.  
    17.     private String db;
    18.     private String table;
    19.  
    20.     private final String NAME="name";
    21.  
    22.     public MetaDataMapFunction(String db, String table) {
    23.         this.db = db;
    24.         this.table = table;
    25.     }
    26.  
    27.     @Override
    28.     public SchemaInfo map(String value) throws Exception {
    29.         String[] aliases = {db.concat(".").concat(table)};
    30.         //存储fields
    31.         HashSet fieldsSet = new HashSet<>();
    32.         //schema支持多种类型，这里我们选择构建常见的record类型
    33.         //初始化结果:
    34.         //{"type":"record","name":"flow","namespace":"com.flow","doc":"fow_event","fields":[],"aliases":["db.table"]}
    35.         SchemaBuilder.RecordBuilder recordBuilder = SchemaBuilder.record("flow").namespace("com.flow").aliases(aliases).doc("fow_event");
    36.         //初始化之后要完善schema中的fields,首先获取fields
    37.         SchemaBuilder.FieldAssembler fields = recordBuilder.fields();
    38.         //处理元数据流
    39.         JSONObject obj = JSON.parseObject(value);
    40.         //过滤掉不包含ID的数据
    41.         Stream data= obj.getJSONArray("data").stream().filter(
    42.                 o -> !StringUtils.isNullOrWhitespaceOnly(JSON.parseObject(o.toString()).getOrDefault(NAME, "").toString())
    43.         ).map(o->JSONObject.parseObject(o.toString()));
    44.         data.forEach(
    45.                 o->{
    46.                     String name = o.get("name").toString();
    47.                     String comment = o.get("comment").toString();
    48.                     buildFields(fields,name,comment);
    49.                     fieldsSet.add(name);
    50.                 }
    51.         );
    52.         Schema schema = fields.endRecord();
    53.         return new SchemaInfo(schema.toString(),fieldsSet);
    54.     }
    55.  
    56.     public void buildFields(SchemaBuilder.FieldAssembler fields,String name,String comment){
    57.         fields.name(name)//字段名称
    58.                 .doc(comment)//描述内容、注释
    59.                 .orderAscending()//排序方式无
    60.                 .type()//类型
    61.                 .optional()
    62.                 .stringType();
    63.     }
    64. }
      65.  

      定义返回对象
       

      1. package bean;
      2.  
      3. import lombok.AllArgsConstructor;
      4. import lombok.Data;
      5.  
      6. import java.util.HashSet;
      7.  
      8.  
      9.  
      10. @Data
      11. @AllArgsConstructor
      12. public class SchemaInfo {
      13.     public String info;
      14.  
      15.     public HashSet set;
      16.  
      17. }
       

      5、延迟处理主流（流量流）
       

      1. package func;
      2.  
      3. import com.alibaba.fastjson.JSON;
      4. import org.apache.flink.api.common.functions.RichMapFunction;
      5. import org.apache.flink.configuration.Configuration;
      6.  
      7. /**
      8.  * 
      9.  * 1、延时map的初始化，阻塞主流数据，等待广播流schema写入广播状态，保证主流数据可以获取到schema
      10.  * 2、标准化数据
      11.  */
      12. public class DelayEtlMap extends RichMapFunction {
      13.  
      14.     private final long delayTime;
      15.  
      16.     public DelayEtlMap(long delayTime) {
      17.         this.delayTime = delayTime;
      18.     }
      19.  
      20.     @Override
      21.     public void open(Configuration parameters) throws Exception {
      22.         super.open(parameters);
      23.         Thread.sleep(delayTime);
      24.     }
      25.  
      26.     @Override
      27.     public String map(String value) throws Exception {
      28.         return value;
      29.     }
      30. }
       

      6、双流关联，处理数据
       

      1. package func;
      2.  
      3. import bean.FlowData;
      4. import bean.SchemaInfo;
      5. import com.alibaba.fastjson.JSON;
      6. import com.alibaba.fastjson.JSONObject;
      7. import org.apache.flink.api.common.state.BroadcastState;
      8. import org.apache.flink.api.common.state.MapStateDescriptor;
      9. import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
      10. import org.apache.flink.api.common.typeinfo.TypeInformation;
      11. import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
      12. import org.apache.flink.util.Collector;
      13.  
      14. import java.text.SimpleDateFormat;
      15. import java.util.HashMap;
      16.  
      17. /**
      18.  * 
      19.  * processElement 处理业务流数据
      20.  * processBroadcastElement 处理广播流数据
      21.  */
      22. public class FlowBrodCastFunction extends BroadcastProcessFunction {
      23.  
      24.     @Override
      25.     public void processElement(String value, BroadcastProcessFunction.ReadOnlyContext ctx, Collector out) throws Exception {
      26.         ReadOnlyBroadcastState flowMetaData = ctx.getBroadcastState(new MapStateDescriptor(
      27.                 "flowMetaData"
      28.                 , TypeInformation.of(String.class)
      29.                 , TypeInformation.of(SchemaInfo.class)));
      30.         SchemaInfo schema = flowMetaData.get("schema");
      31.         JSONObject jsonObject = JSON.parseObject(value);
      32.         Long time = jsonObject.getLong("time");
      33.         //创建时间格式
      34.         SimpleDateFormat yyyyMMdd = new SimpleDateFormat("yyyyMMdd");
      35.         String dt = yyyyMMdd.format(time);
      36.         //获取元数据字段
      37.         HashMap map = new HashMap<>();
      38.         schema.getSet().forEach(
      39.                 o->{
      40.                     map.put(o.toString(),jsonObject.get(o).toString());
      41.                 }
      42.         );
      43.         out.collect(new FlowData(schema.getInfo(),map,time));
      44.     }
      45.  
      46.     @Override
      47.     public void processBroadcastElement(SchemaInfo value, BroadcastProcessFunction.Context ctx, Collector out) throws Exception {
      48.         if(value != null){
      49.             BroadcastState flowMetaData = ctx.getBroadcastState(new MapStateDescriptor(
      50.                     "flowMetaData"
      51.                     , TypeInformation.of(String.class)
      52.                     , TypeInformation.of(SchemaInfo.class)));
      53.             flowMetaData.put("schema", value);
      54.         }
      55.     }
      56. }
       

      FlowData对象
       

      1. package bean;
      2.  
      3. import lombok.AllArgsConstructor;
      4. import lombok.Data;
      5.  
      6. import java.util.HashMap;
      7.  
      8.  
      9. @Data
      10. @AllArgsConstructor
      11. public class FlowData {
      12.     public String schema;
      13.  
      14.     public HashMap values;
      15.  
      16.     private long time;
      17. }
       

      7、主代码
       

      1. package ubs.app;
      2.  
      3. import bean.FlowData;
      4. import bean.SchemaInfo;
      5. import func.DelayEtlMap;
      6. import func.FlowBrodCastFunction;
      7. import func.FlowSinkFunction;
      8. import func.MetaDataMapFunction;
      9. import org.apache.flink.api.common.state.MapStateDescriptor;
      10. import org.apache.flink.api.common.typeinfo.TypeInformation;
      11. import org.apache.flink.api.java.utils.ParameterTool;
      12. import org.apache.flink.streaming.api.datastream.BroadcastStream;
      13. import org.apache.flink.streaming.api.datastream.DataStreamSource;
      14. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
      15. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
      16. import org.apache.flink.util.StringUtils;
      17. import source.FlowSourceFunction;
      18. import source.StringSource;
      19.  
      20. import java.util.Objects;
      21.  
      22.  
      23. public class FlowApp {
      24.     public static void main(String[] args) throws Exception{
      25.         //解析参数
      26.         ParameterTool parameterTool = ParameterTool.fromArgs(args);
      27.         //获取参数 db
      28.         String db = parameterTool.get("db")==null?"":parameterTool.get("db");
      29.         //获取参数 table
      30.         String table = parameterTool.get("table")==null?"":parameterTool.get("table");
      31.         //初始化环境
      32.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      33.         //模拟元数据流，后续获取到元数据进行广播
      34.         DataStreamSource configSource = env.addSource(new StringSource());
      35.         //设置并行度
      36.         configSource.setParallelism(1);
      37.         //获取实时流量数据流:kafka source
      38. //        KafkaSource source = SourceGetter.getValueOnlySimpleStrDesSource(parameterTool);
      39. //        DataStreamSource realData = env.fromSource(source, WatermarkStrategy.noWatermarks(), "liuliang");
      40.         DataStreamSource realData = env.addSource(new FlowSourceFunction());
      41.         realData.print("realData: ");
      42. //        //处理元数据并广播：将元数据流接收并处理为schema
      43.         BroadcastStream metaData = configSource.filter(data -> !StringUtils.isNullOrWhitespaceOnly(data))
      44.                 .setParallelism(1)
      45.                 .map(new MetaDataMapFunction("a", "b"))
      46.                 .broadcast(new MapStateDescriptor(
      47.                         "flowMetaData"
      48.                         , TypeInformation.of(String.class)
      49.                         , TypeInformation.of(SchemaInfo.class)
      50.                 ));
      51.         //延迟主流
      52.         SingleOutputStreamOperator realData2 = realData.map(new DelayEtlMap(1000))
      53.                 .filter(Objects::nonNull);
      54.         //连接流，解析数据
      55.         SingleOutputStreamOperator res = realData2.connect(metaData).process(new FlowBrodCastFunction());
      56.         res.addSink(new FlowSinkFunction());
      57.         res.print("res");
      58.         env.execute();
      59.     }
      60. }
       

      
                
                    
                

                
    65. 

                    相关阅读:
      
                    
消息队列MQ核心原理全面总结(11大必会原理)                            
      
Java学习 - MySQL数据库中提到的 视图 是什么？ 如何使用？                            
      
【案例30】WebSphere诡异宕机                            
      
mysql foreach 写法，多条 sql 与一条 sql 的性能消耗分析                            
      
Java对List的操作                            
      
【计算机网络——应用层】http协议                            
      
【Arduino25】液晶模拟值实验                            
      
Python重点数据结构基本用法                            
      
【英雄哥七月集训】第 26天：并查集                            
      
xshell不能通过账户密码连接虚拟机                            
      
                          
                
      66. 
                
    66. 
                    原文地址：https://blog.csdn.net/qq_40771567/article/details/136379911
                
      67.