flink bitmap去重并写入RabitMQ

大数据启示录 2022-02-15

147

1:样例数据集

{"uuid": "7203d5f8a55c11ea9986a64728966615", "title": "暴乱持续发酵，特朗普3小时发16条推文，疫苗又有新进展？惊现巨量抛单，黄金闪跌20美元 | 早报", "url": "http://www.toutiao.com/item/6833955164274033159/","ctime": 1590019056, "utime": 0} 
{"uuid": "43f67f489aef11ea967c8e469741ac9c", "title": "汇添富成长多因子量化策略股票净值下跌1.55% 请保持关注", "url": "http://www.toutiao.com/item/6829065074540282376/","ctime": 1590019056, "utime": 0}                                     
{"uuid": "c44f94129aff11ea967c8e469741ac9c", "title": "上投摩根内需动力混合基金最新净值跌幅达1.66%", "url": "http://www.toutiao.com/item/6829079830701539847/","ctime": 1590019056, "utime": 0 }                                             
{"uuid": "c50c65929aff11ea967c8e469741ac9c", "title": "银华体育文化灵活配置混合基金最新净值跌幅达1.68%", "url": "http://www.toutiao.com/item/6829079830709928455/","ctime": 1590019056, "utime": 0}                                          
{"uuid": "2f18d7cea05d11ea9980a64728966615", "title": "68家上市游戏公司Q1财报梳理：36家净利润下滑，谁吃到了宅经济红利？", "url": "http://www.toutiao.com/item/6831618549249737229/","ctime": 1590019056, "utime": 0}                         
{"uuid": "c78a036a9aff11ea967c8e469741ac9c", "title": "长城中小盘成长混合基金最新净值跌幅达2.45%", "url": "http://www.toutiao.com/item/6829079832752554510/","ctime": 1590019056, "utime": 0}                                               
{"uuid": "602d7c2aa07111ea96878e469741ac9c", "title": "东方量化成长灵活配置混合基金最新净值跌幅达1.75%", "url": "http://www.toutiao.com/item/6831661455230632461/","ctime": 1590019056, "utime": 0}                                          
{"uuid": "c37db9b09aff11ea967c8e469741ac9c", "title": "融通蓝筹成长混合基金最新净值跌幅达1.66%", "url": "http://www.toutiao.com/item/6829079847638139399/","ctime": 1590019056, "utime": 0}                                                 
{"uuid": "5f0d2444a07111ea9980a64728966615", "title": "银华瑞和灵活配置混合基金最新净值跌幅达1.94%", "url": "http://www.toutiao.com/item/6831661456606364167/","ctime": 1590019056, "utime": 0}                                              
{"uuid": "65d4ba4c9b0011ea9972a64728966615", "title": "“宅经济”盛行 游戏上市公司一季度“成绩”逆势上涨", "url": "http://www.toutiao.com/item/6829086094345110029/","ctime": 1590019056, "utime": 0}                                            
{"uuid": "7203d5f8a55c11ea9986a64728966615", "title": "暴乱持续发酵，特朗普3小时发16条推文，疫苗又有新进展？惊现巨量抛单，黄金闪跌20美元 | 早报", "url": "http://www.toutiao.com/item/6833955164274033159/","ctime": 1590019056, "utime": 0} 
{"uuid": "43f67f489aef11ea967c8e469741ac9c", "title": "汇添富成长多因子量化策略股票净值下跌1.55% 请保持关注", "url": "http://www.toutiao.com/item/6829065074540282376/","ctime": 1590019056, "utime": 0}                                     
{"uuid": "c44f94129aff11ea967c8e469741ac9c", "title": "上投摩根内需动力混合基金最新净值跌幅达1.66%", "url": "http://www.toutiao.com/item/6829079830701539847/","ctime": 1590019056, "utime": 0 }                                             
{"uuid": "c50c65929aff11ea967c8e469741ac9c", "title": "银华体育文化灵活配置混合基金最新净值跌幅达1.68%", "url": "http://www.toutiao.com/item/6829079830709928455/","ctime": 1590019056, "utime": 0}                                          
{"uuid": "2f18d7cea05d11ea9980a64728966615", "title": "68家上市游戏公司Q1财报梳理：36家净利润下滑，谁吃到了宅经济红利？", "url": "http://www.toutiao.com/item/6831618549249737229/","ctime": 1590019056, "utime": 0}                         
{"uuid": "c78a036a9aff11ea967c8e469741ac9c", "title": "长城中小盘成长混合基金最新净值跌幅达2.45%", "url": "http://www.toutiao.com/item/6829079832752554510/","ctime": 1590019056, "utime": 0}                                               
{"uuid": "602d7c2aa07111ea96878e469741ac9c", "title": "东方量化成长灵活配置混合基金最新净值跌幅达1.75%", "url": "http://www.toutiao.com/item/6831661455230632461/","ctime": 1590019056, "utime": 0}                                          
{"uuid": "c37db9b09aff11ea967c8e469741ac9c", "title": "融通蓝筹成长混合基金最新净值跌幅达1.66%", "url": "http://www.toutiao.com/item/6829079847638139399/","ctime": 1590019056, "utime": 0}                                                 
{"uuid": "5f0d2444a07111ea9980a64728966615", "title": "银华瑞和灵活配置混合基金最新净值跌幅达1.94%", "url": "http://www.toutiao.com/item/6831661456606364167/","ctime": 1590019056, "utime": 0}                                              
{"uuid": "65d4ba4c9b0011ea9972a64728966615", "title": "“宅经济”盛行 游戏上市公司一季度“成绩”逆势上涨", "url": "http://www.toutiao.com/item/6829086094345110029/","ctime": 1590019056, "utime": 0}                                            
{"uuid": "7203d5f8a55c11ea9986a64728966615", "title": "暴乱持续发酵，特朗普3小时发16条推文，疫苗又有新进展？惊现巨量抛单，黄金闪跌20美元 | 早报", "url": "http://www.toutiao.com/item/6833955164274033159/","ctime": 1590019056, "utime": 0} 
{"uuid": "43f67f489aef11ea967c8e469741ac9c", "title": "汇添富成长多因子量化策略股票净值下跌1.55% 请保持关注", "url": "http://www.toutiao.com/item/6829065074540282376/","ctime": 1590019056, "utime": 0}                                     
{"uuid": "c44f94129aff11ea967c8e469741ac9c", "title": "上投摩根内需动力混合基金最新净值跌幅达1.66%", "url": "http://www.toutiao.com/item/6829079830701539847/","ctime": 1590019056, "utime": 0 }                                             
{"uuid": "c50c65929aff11ea967c8e469741ac9c", "title": "银华体育文化灵活配置混合基金最新净值跌幅达1.68%", "url": "http://www.toutiao.com/item/6829079830709928455/","ctime": 1590019056, "utime": 0}                                          
{"uuid": "2f18d7cea05d11ea9980a64728966615", "title": "68家上市游戏公司Q1财报梳理：36家净利润下滑，谁吃到了宅经济红利？", "url": "http://www.toutiao.com/item/6831618549249737229/","ctime": 1590019056, "utime": 0}                         
{"uuid": "c78a036a9aff11ea967c8e469741ac9c", "title": "长城中小盘成长混合基金最新净值跌幅达2.45%", "url": "http://www.toutiao.com/item/6829079832752554510/","ctime": 1590019056, "utime": 0}                                               
{"uuid": "602d7c2aa07111ea96878e469741ac9c", "title": "东方量化成长灵活配置混合基金最新净值跌幅达1.75%", "url": "http://www.toutiao.com/item/6831661455230632461/","ctime": 1590019056, "utime": 0}                                          
{"uuid": "c37db9b09aff11ea967c8e469741ac9c", "title": "融通蓝筹成长混合基金最新净值跌幅达1.66%", "url": "http://www.toutiao.com/item/6829079847638139399/","ctime": 1590019056, "utime": 0}                                                 
{"uuid": "5f0d2444a07111ea9980a64728966615", "title": "银华瑞和灵活配置混合基金最新净值跌幅达1.94%", "url": "http://www.toutiao.com/item/6831661456606364167/","ctime": 1590019056, "utime": 0}                                              
{"uuid": "65d4ba4c9b0011ea9972a64728966615", "title": "“宅经济”盛行 游戏上市公司一季度“成绩”逆势上涨", "url": "http://www.toutiao.com/item/6829086094345110029/","ctime": 1590019056, "utime": 0}                                            
{"uuid": "7203d5f8a55c11ea9986a64728966615", "title": "暴乱持续发酵，特朗普3小时发16条推文，疫苗又有新进展？惊现巨量抛单，黄金闪跌20美元 | 早报", "url": "http://www.toutiao.com/item/6833955164274033159/","ctime": 1590019056, "utime": 0} 
{"uuid": "43f67f489aef11ea967c8e469741ac9c", "title": "汇添富成长多因子量化策略股票净值下跌1.55% 请保持关注", "url": "http://www.toutiao.com/item/6829065074540282376/","ctime": 1590019056, "utime": 0}                                     
{"uuid": "c44f94129aff11ea967c8e469741ac9c", "title": "上投摩根内需动力混合基金最新净值跌幅达1.66%", "url": "http://www.toutiao.com/item/6829079830701539847/","ctime": 1590019056, "utime": 0 }                                             
{"uuid": "c50c65929aff11ea967c8e469741ac9c", "title": "银华体育文化灵活配置混合基金最新净值跌幅达1.68%", "url": "http://www.toutiao.com/item/6829079830709928455/","ctime": 1590019056, "utime": 0}                                          
{"uuid": "2f18d7cea05d11ea9980a64728966615", "title": "68家上市游戏公司Q1财报梳理：36家净利润下滑，谁吃到了宅经济红利？", "url": "http://www.toutiao.com/item/6831618549249737229/","ctime": 1590019056, "utime": 0}                         
{"uuid": "c78a036a9aff11ea967c8e469741ac9c", "title": "长城中小盘成长混合基金最新净值跌幅达2.45%", "url": "http://www.toutiao.com/item/6829079832752554510/","ctime": 1590019056, "utime": 0}                                               
{"uuid": "602d7c2aa07111ea96878e469741ac9c", "title": "东方量化成长灵活配置混合基金最新净值跌幅达1.75%", "url": "http://www.toutiao.com/item/6831661455230632461/","ctime": 1590019056, "utime": 0}                                          
{"uuid": "c37db9b09aff11ea967c8e469741ac9c", "title": "融通蓝筹成长混合基金最新净值跌幅达1.66%", "url": "http://www.toutiao.com/item/6829079847638139399/","ctime": 1590019056, "utime": 0}                                                 
{"uuid": "5f0d2444a07111ea9980a64728966615", "title": "银华瑞和灵活配置混合基金最新净值跌幅达1.94%", "url": "http://www.toutiao.com/item/6831661456606364167/","ctime": 1590019056, "utime": 0}                                              
{"uuid": "65d4ba4c9b0011ea9972a64728966615", "title": "“宅经济”盛行 游戏上市公司一季度“成绩”逆势上涨", "url": "http://www.toutiao.com/item/6829086094345110029/","ctime": 1590019056, "utime": 0}

‍

数据说明：此样例数据共有40条，其中只有10条数据uuid不重复。

2: 依赖导入

 <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-rabbitmq -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-rabbitmq_2.11</artifactId>
            <version>1.12.2</version>
            <scope>provided</scope>
        </dependency>

3: 核心代码

3.1 读取样例数据，并解析。通过ValueState结合bitmap 进行去重

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple10;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.util.Collector;


public class bitMap2MQ {
    public static void main(String[] args) throws Exception {
        //创建环境，设置参数
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //读取数据源
        DataStreamSource<String> pulsarDS = env.readTextFile("/Users/mindezhi/Downloads/PublicSentiment/src/main/data/test2.txt");
        ObjectMapper mapper = new ObjectMapper();


        SingleOutputStreamOperator<Tuple5<String, String, String, String, String>> mapDS = pulsarDS.map(new MapFunction<String, Tuple5<String, String, String,String,String>>() {
            @Override
            public Tuple5<String, String, String, String, String> map(String value) throws Exception {
                JsonNode jsonNode = mapper.readTree(value);
                String uuid = (jsonNode.get("uuid") != null) ? jsonNode.get("uuid").toString() : "";
                String ctime = (jsonNode.get("ctime") != null) ? jsonNode.get("ctime").toString() : "";
                String utime = (jsonNode.get("utime") != null) ? jsonNode.get("utime").toString() : "";
                String title = (jsonNode.get("title") != null) ? jsonNode.get("title").toString() : "";
                String url = (jsonNode.get("url") != null) ? jsonNode.get("url").toString() : "";
                return Tuple5.of(uuid, ctime, utime, title, url);
            }
        });
        KeyedStream<Tuple5<String, String, String, String, String>, Tuple> KeyedStream = mapDS.keyBy(1);


        final SingleOutputStreamOperator<Tuple5<String, String, String, String,String>> distinctProcess = KeyedStream.process(new KeyedProcessFunction<Tuple, Tuple5<String, String, String, String, String>, Tuple5<String, String, String, String, String>>() {
            // 定义一个状态，用来存放BloomFilter
            private transient ValueState<BloomFilter> uidState;
            @Override
            public void open(Configuration parameters) throws Exception {
                // 定义一个状态描述器 保存BloomFilter
                ValueStateDescriptor<BloomFilter> stateDescriptor1 = new ValueStateDescriptor(
                        "uid-stated",
                        TypeInformation.of(new TypeHint<BloomFilter>() {
                        })
                );
                // 获取历史状态
                uidState = getRuntimeContext().getState(stateDescriptor1);
            }


            @Override
            public void processElement(Tuple5<String,String, String, String, String> value, Context ctx, Collector<Tuple10<String, String, String, String, String, String, String, String, String, String>> out) throws Exception {
                String uuid = value.f0;
                String ctime = value.f1;
                String utime = value.f2;
                String title = value.f3;
                String url = value.f4;
               
                // 使用BloomFilter判断去重
                BloomFilter bloomFilter = uidState.value();


                if (bloomFilter == null) {
                    bloomFilter = BloomFilter.create(Funnels.unencodedCharsFunnel(), 10000000);
                }
                // 如果数据不存在，则将数据添加到bloomFilter中
                if (!bloomFilter.mightContain(uuid)&&!bloomFilter.mightContain(title)) {
                    bloomFilter.put(uuid);
                    uidState.update(bloomFilter);
                    out.collect(Tuple5.of(uuid, ctime, utime, title, url));
                }


            }
        });

4: 定义rabbitMQ sink,由于flink-connector-rabbitmq 发送数据是以string类型发送，在此需要，将数据封装成json形式的字符串。

         // 封装成json形式的String        
        SingleOutputStreamOperator<String> mapDs = distinctProcess.map(new MapFunction<Tuple5<String, String, String, String, String>, String>() {
            @Override
            public String map(Tuple5< String, String, String, String, String> value) throws Exception {
                String str = "{ " +
                        "\"msg_header\":{" +
                        "\"uuid\":" + value.f0 + "," +
                        "\"ctime\":" + value.f1 + "," +
                        "\"utime\":" + value.f2 + "," +
                        "\"duplicate_filter\":0," +
                        "\"duplicate\":0}," +
                        "\"msg_body\":{" +
                        "\"title\":" + value.f3 + "," +
                        "\"stocks\":[]," +
                        "\"url\":" + value.f4 + "," +
                        "\"tags\":[]" +
                        "}}";


                return str;
            }
        });




        RMQConnectionConfig connectionConfig = new RMQConnectionConfig
                .Builder().setHost("localhost").setVirtualHost("/")
                .setPort(5672).setUserName("rabbitMQ").setPassword("xxx@2020")
                .build();


        //注意，换一个新的 queue，否则也会报错
        mapDs.addSink(new RMQSink(connectionConfig, "disinct_test",new SimpleStringSchema()));
        env.execute();


    }
}

数据库

文章转载自大数据启示录，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

flink bitmap去重并写入RabitMQ

评论