高考报名时间2022年11月1日还剩 天 距离2023年6月7日高考还有 天
栏目分类:
子分类:
返回
名师互学网用户登录
名师互学网
快速导航关闭
当前搜索
当前分类
前沿技术 软件开发 系统运维 产品运营 生活办公 面试经验 考试题库
子分类
人工智能 大数据 云计算 区块链 物联网 深度学习 机器学习 NLP 计算机视觉 语音识别 其他 大数据系统 数据可视化 数据挖掘与分析 其他 Docker/k8s 虚拟化 云平台 其他 基本原理 数字货币 智能合约 EOS应用 其他 通讯技术 嵌入式开发 单片机 物联网应用 HarmonyOS 其他 后端开发 Web开发 移动开发 游戏开发 Python Java 架构设计 C/C++/C# PHP .Net Go语言 R语言
实用工具
学习工具 小学数学练习 字帖生成 在线画板 函数绘制 拼音字母表 在线词典 黄历查询 亲戚关系计算 安全期计算 中国历史 Excel函数 模拟请求 json格式化 浏览器指纹
热门搜索
路由器设置 木托盘 宝塔面板 儿童python教程 心情低落 朋友圈 vim 双一流学科 专升本 我的学校 日记学校 西点培训学校 汽修学校 情书 化妆学校 塔沟武校 异形模板 西南大学排名 最精辟人生短句 6步教你追回被骗的钱 南昌大学排名 清朝十二帝 北京印刷学院排名 北方工业大学排名 北京航空航天大学排名 首都经济贸易大学排名 中国传媒大学排名 首都师范大学排名 中国地质大学(北京)排名 北京信息科技大学排名
名师互学网 > IT > 前沿技术 > 大数据 > 大数据系统

Flink旁路输出特性简单实例：按照股价对股票进行数据分流并写出到文本文件

大数据系统 更新时间： 发布时间： IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

Flink旁路输出特性简单实例：按照股价对股票进行数据分流并写出到文本文件

关于旁路输出的官方文档：https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/dev/datastream/side_output/

除了由 DataStream 操作产生的主要流之外，我们还可以产生任意数量的旁路输出结果流。结果流中的数据类型不必与主要流中的数据类型相匹配，并且不同旁路输出的类型也可以不同。

使用旁路输出时，首先需要定义用于标识旁路输出流的OutputTag类对象。
构造方法的第一个参数表示一个区分旁路输出流的id标识，第二个参数表示要处理的数据类型。

OutputTag outputTag = new OutputTag("side-output", Types.STRING);

定义旁路输出标签后，通过主输出流的process方法，把数据发送到旁路输出流中。

SingleOutputStreamOperator process = input
  .process(new ProcessFunction() {

      @Override
      public void processElement(
          String value,
          Context ctx,
          Collector collector) throws Exception {
        // 发送数据到主要的输出
        collector.collect(value);
        // 发送数据到旁路输出
        ctx.output(outputTag, "sideout-" + value);
      }
    });
 

我们可以在 DataStream 运算结果上使用 getSideOutput(OutputTag) 方法获取旁路输出流，这将产生一个与旁路输出流结果类型一致的 DataStream。
 

DataStream sideOutputStream = mainDataStream.getSideOutput(outputTag);

 

 

以股票为例，在发送股票数据时，我们假设股价小于50就是低价股，否则就是高级股，在发送股票数据时，我们希望把高价股和低价股分别写出到不同的文件中保存起来。
 

首先，创建一个股票类Stock：
 

public class Stock {
	//股票名称
	private String name;
	//股票价格
	private Integer price;
	//构造方法、getter、setter、toString方法在此省略
}

 

其次，编写Flink消费者程序：
 

public class FlinkKafkaConsumer{

	public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.配置文件
        Properties props = new Properties();
        props.put("bootstrap.servers","Kafka集群地址");
        //3.构造消费者
        FlinkKafkaConsumer consumer = new FlinkKafkaConsumer<>("stock", new SimpleStringSchema(), props);
        //4.配置消费者
        DataStreamSource stream = env.addSource(consumer);
        //5.data sick
        stream.addSick(new SinkFunction(){
			@Override
			public void invoke(String value, Context context){
				Syestem.out.println("当前已处理的数据：" + JsonUtils.deserialize(value, Stock.class));
			}
		});
        //6.执行程序
        env.execute("消费者程序");
    }
}


 

接着，编写生产者程序：
 

​public class FlinkProducer {

    public static void main(String[] args) throws Exception {
        //结果输出路径
        String outputPath1 = "C:\Users\xxx\Desktop\result1";
        String outputPath2 = "C:\Users\xxx\Desktop\result2";
        //创建数据源
        ArrayList stocks = new ArrayList<>();
        for (int i = 0; i < 100 ; i++) {
            stocks.add(JsonUtils.serialize(new Stock("stock-" + i, (int)(Math.random() * 100))));
        }
        //创建消费者环境
        StreamExecutionEnvironment producerEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        //配置文件
        Properties props = new Properties();
        props.put("bootstrap.servers", "10.225.173.107:9092,10.225.173.108:9092,10.225.173.109:9092");
        //构建生产者
        FlinkKafkaProducer producer_all = new FlinkKafkaProducer<>("STOCK", new SimpleStringSchema(), props);
        //配置数据源
        DataStreamSource stream_all = producerEnvironment.fromCollection(stocks);
        //创建旁路输出
        final OutputTag outputTag_lowPrice = new OutputTag<>("STOCK_LOW_PRICE", Types.STRING);
        final OutputTag outputTag_highPrice = new OutputTag<>("STOCK_HIGH_PRICE", Types.STRING);
        //配置旁路输出
        SingleOutputStreamOperator process = stream_all.process(new ProcessFunction() {
            @Override
            public void processElement(String s, ProcessFunction.Context context, Collector collector) {
                Stock stock = JsonUtils.deserialize(s, Stock.class);
                collector.collect(s);
                if (stock.getPrice() < 50) {
                    context.output(outputTag_lowPrice, s);
                } else {
                    context.output(outputTag_highPrice, s);
                }
            }
        });
        //获取低价股票和高价股票的旁路输出
        DataStream stream_lowPrice = process.getSideOutput(outputTag_lowPrice);
        DataStream stream_highPrice = process.getSideOutput(outputTag_highPrice);
        //配置生产者
        stream_all.addSink(producer_all);
        //处理低价股票
        stream_lowPrice.writeAsText(outputPath1, FileSystem.WriteMode.OVERWRITE);
        //处理高价股票
        stream_highPrice.writeAsText(outputPath2, FileSystem.WriteMode.OVERWRITE);
        //执行flink程序
        producerEnvironment.execute("生产者流处理");
    }

}
 

依次启动消费者程序、生产者程序，观察消费者程序控制台中的输出：
 
 此时，桌面生成了两个文件夹，result1记录了小于50的股票，result2相反：
 







转载请注明：文章转载自 www.mshxw.com















大数据系统相关栏目本月热门文章












热门相关搜索






路由器设置
木托盘
宝塔面板
儿童python教程
心情低落
朋友圈
vim
双一流学科
专升本
我的学校
日记学校
西点培训学校
汽修学校
情书
化妆学校
塔沟武校
异形模板
西南大学排名
最精辟人生短句
6步教你追回被骗的钱
南昌大学排名
清朝十二帝
北京印刷学院排名
北方工业大学排名
北京航空航天大学排名
首都经济贸易大学排名
中国传媒大学排名
首都师范大学排名
中国地质大学(北京)排名
北京信息科技大学排名
中央民族大学排名
北京舞蹈学院排名
北京电影学院排名
中国戏曲学院排名
河北政法职业学院排名
河北经贸大学排名
天津中德应用技术大学排名
天津医学高等专科学校排名
天津美术学院排名
天津音乐学院排名
天津工业大学排名
北京工业大学耿丹学院排名
北京警察学院排名
天津科技大学排名
北京邮电大学(宏福校区)排名
北京网络职业学院排名
北京大学医学部排名
河北科技大学排名
河北地质大学排名
河北体育学院排名















学习工具


代数计算器


三角函数


解析几何


立体几何






知识解答


教育知识


百科知识


生活知识


常识知识






写作必备


作文大全


作文素材


句子大全



实用范文






关于我们


关于我们


联系我们


网站地图






 交流群

名师互学网交流群




名师互学网客服

名师互学网客服












名师互学网 版权所有 (c)2021-2022      ICP备案号：晋ICP备2021003244-6号
 








我们一直用心在做

关于我们
文章归档
网站地图
联系我们

版权所有 (c)2021-2022 MSHXW.COM


ICP备案号：晋ICP备2021003244-6号