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(); } }