Java版Flink使用指南——定制RabbitMQ数据源的序列化器

大纲

  • 新建工程
    • 新增依赖
    • 数据对象
    • 序列化器
    • 接入数据源
  • 测试
    • 修改Slot个数
    • 打包、提交、运行
  • 工程代码

在《Java版Flink使用指南——从RabbitMQ中队列中接入消息流》一文中,我们从RabbitMQ队列中读取了字符串型数据。如果我们希望读取的数据被自动化转换为一个对象,则需要定制序列化器。本文我们就将讲解数据源序列化器的定制方法。

新建工程

我们在IntelliJ中新建一个工程SourceSerializer。
Archetype填入:org.apache.flink:flink-quickstart-java
版本填入与Flink的版本:1.19.1
在这里插入图片描述

新增依赖

在pom.xml中新增RabbitMQ连接器

		<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-rabbitmq</artifactId><version>3.0.1-1.17</version></dependency>

新增Json库依赖

		<dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-core</artifactId><version>2.17.1</version></dependency>

新增lombok库,主要是为了使用它的一些注解

        <dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.32</version><scope>provided</scope></dependency>

数据对象

我们新建一个简单的数据对象SampleData
src/main/java/org/example/vo/SampleData.java

package org.example.vo;import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;@Data
@NoArgsConstructor
@AllArgsConstructor
public class SampleData {private Long id;private String name;private int age;private Boolean married;private Double salary;public String toJson() throws JsonProcessingException {ObjectMapper mapper = new ObjectMapper();return mapper.writeValueAsString(this);}public static SampleData fromJson(String json) throws JsonProcessingException {ObjectMapper mapper = new ObjectMapper();return mapper.readValue(json, SampleData.class);}
}

这个方法包含两个方法,一个是将SampleData 转换成字符串,另一个是将字符串转成SampleData 对象。

序列化器

我们定义的数据源序列化器要实现AbstractDeserializationSchema接口,主要是通过deserialize方法将二进制数组转换成SampleData 对象。

src/main/java/org/example/serializer/SampleDataRabbitMQSourceSerializer.java

package org.example.serializer;import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.example.vo.SampleData;import java.io.IOException;public class SampleDataRabbitMQSourceSerializer extends AbstractDeserializationSchema<SampleData> {@Overridepublic SampleData deserialize(byte[] message) throws IOException {return SampleData.fromJson(new String(message));}@Overridepublic boolean isEndOfStream(SampleData nextElement) {return false;}@Overridepublic TypeInformation<SampleData> getProducedType() {return TypeInformation.of(SampleData.class);}
}

接入数据源

我们在《Java版Flink使用指南——定制RabbitMQ的Sink序列化器》一文中,往data.to.rbtmq对了写入了大量SampleData 数据。这次我们将其作为数据源来做测试
这次我们在创建RMQSource时传入序列化器SampleDataRabbitMQSourceSerializer。它会将从RabbitMQ获取的数据转换成SampleData对象。
然后我们获取所有“已婚”(filter.getMarried() == true)的数据,将其打印到日志中。

		String queueName = "data.to.rbtmq";String host = "172.21.112.140"; // IP of the rabbitmq serverint port = 5672;String username = "admin";String password = "fangliang";String virtualHost = "/";int parallelism = 1;// create a RabbitMQ sourceRMQConnectionConfig rmqConnectionConfig = new RMQConnectionConfig.Builder().setHost(host).setPort(port).setUserName(username).setPassword(password).setVirtualHost(virtualHost).build();RMQSource<SampleData> rmqSource = new RMQSource<>(rmqConnectionConfig, queueName, true, new SampleDataRabbitMQSourceSerializer());final DataStream<SampleData> stream = env.addSource(rmqSource).name(username + "'s source from " + queueName).setParallelism(parallelism);stream.filter(filter -> filter.getMarried() == true).print().name(username + "'s sink to stdout").setParallelism(parallelism);

测试

修改Slot个数

由于我们要运行两个流式计算任务,于是需要两个Slot。

vim conf/config.yaml 

将numberOfTaskSlots的值改成2。

打包、提交、运行

我们将本例和《Java版Flink使用指南——定制RabbitMQ的Sink序列化器》中的包都提交运行
在这里插入图片描述
然后在日志中可以看到“已婚”的数据都在输出

 tail -f log/*

在这里插入图片描述

工程代码

https://github.com/f304646673/FlinkDemo

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/diannao/43286.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Linux C++ 043-机房预约系统

Linux C 043-机房预约系统 本节关键字&#xff1a;Linux、C、机房预约系统 相关库函数&#xff1a;for_each、open、close、write 系统简介 学校现在有几个规格不同的机房&#xff0c;由于使用经常出现撞车现象&#xff0c;现开发一套机房预约系统&#xff0c;解决这一问题。…

Java进阶---抽象方法abstract

抽象方法 案例引入: 在某个宠物店的宠物资源管理系统中有&#xff1a; 狗类&#xff1a;属性&#xff08;姓名&#xff09;&#xff0c;行为&#xff08;吃饭&#xff09; 猫类&#xff1a;属性&#xff08;姓名&#xff09;&#xff0c;行为&#xff08;吃饭&#xff09;利用…

智慧科技照亮水利未来:深入剖析智慧水利解决方案如何助力水利行业实现高效、精准、可持续的管理

目录 一、智慧水利的概念与内涵 二、智慧水利解决方案的核心要素 1. 物联网技术&#xff1a;构建全面感知网络 2. 大数据与云计算&#xff1a;实现数据高效处理与存储 3. GIS与三维可视化&#xff1a;提升决策支持能力 4. 人工智能与机器学习&#xff1a;驱动决策智能化 …

LibreOffice的国内镜像安装地址和node.js国内快速下载网站

文章目录 1、LibreOffice1.1、LibreOffice在application-conf.yml中的配置2、node.js 1、LibreOffice 国内镜像包网址&#xff1a;https://mirrors.cloud.tencent.com/libreoffice/libreoffice/ 1.1、LibreOffice在application-conf.yml中的配置 jodconverter:local:enable…

Java面试八股之MySQL中int(10)和bigint(10)能存储读的数据大小一样吗

MySQL中int(10)和bigint(10)能存储读的数据大小一样吗 在MySQL中&#xff0c;int(10)和bigint(10)的数据存储能力并不相同&#xff0c;尽管括号内的数字&#xff08;如10&#xff09;看起来似乎暗示着某种关联&#xff0c;但实际上这个数字代表的是显示宽度&#xff0c;而不是…

vue学习day03-指令修饰符、v-bind对于样式控制的增强、v-model应用于其他表单元素

7、指令修饰符 &#xff08;1&#xff09;概念&#xff1a; 通过“.”指明一些指令后缀&#xff0c;不同后缀封装了不同的处理操作->简化代码 &#xff08;2&#xff09;按键修饰符 keyup.enter->键盘回车监听 &#xff08;3&#xff09;v-model修饰符 v-model.tri…

vue + element ui 实现侧边栏导航栏折叠收起

首页布局如下 要求点击按钮,将侧边栏收缩, 通过 row 和 col 组件&#xff0c;并通过 col 组件的 span 属性我们就可以自由地组合布局。 折叠前 折叠后 <template><div class"app-layout" :class"{ collapse: app.isFold }"><div class&…

Onekey正版steam分流下载工具

今天给大家介绍的是一款下载steam游戏的工具。Onekey工具&#xff0c;是一款游戏下载器&#xff0c;可以下载steam正版分流游戏。下载正版分流的网站很多&#xff0c;但是都是网盘或者迅雷下载&#xff0c;或者游戏盒子下载&#xff0c;速度都很慢。这款软件是用steam下载的&am…

Flask项目搭建及部署 —— Python

flask搭建及部署 pip 19.2.3 python 3.7.5 Flask 1.1.1 Flask-SQLAlchemy 2.4.1 Pika 1.1.0 Redis 3.3.11 flask-wtf 0.14.2 1、创建flask项目&#xff1a; 创建完成后整个项目结构树&#xff1a; app.py: 项⽬管理⽂件&#xff0c;通过它管理项⽬。 static: 存放静态…

自定义控件视图篇(一)测量与布局

在自定义控件的开发过程中&#xff0c;"视图篇"的测量与布局是非常关键的步骤&#xff0c;这直接决定了控件的尺寸、位置以及子视图的排列方式。下面我将详细介绍测量和布局的过程&#xff0c;以及如何在自定义控件中正确实现这些步骤。 视图的测量 (onMeasure) 在…

2021版本的idea热部署的详细步骤

背景&#xff1a;我是自己用的是2021版本的idea,然后发现跟2023版本的热部署不太一样&#xff0c;所以&#xff0c;今天自己出一期这样的文章吧&#xff01;&#xff01;&#xff01;其他人配置的时候根据自己的情况&#xff0c;来阅读吧&#xff01; 第一步&#xff1a;方式一…

MyBatis是如何分页的及原理

MyBatis 是一种持久层框架&#xff0c;支持通过配置文件和注解将 SQL 映射为 Java 对象。在实际开发中&#xff0c;查询数据时经常需要进行分页处理。 MyBatis 也提供了支持分页的方案&#xff0c;其主要思路是使用 Limit 偏移量和限制个数&#xff0c;来获取指定数量的数据。下…

音视频入门基础:H.264专题(10)——FFmpeg源码中,存放SPS属性的结构体和解码SPS的函数分析

一、引言 FFmpeg源码对AnnexB包装的H.264码流解码过程中&#xff0c;通过ff_h2645_extract_rbsp函数拿到该H.264码流中的某个NALU的NALU Header RBSP后&#xff08;具体可以参考&#xff1a;《FFmpeg源码&#xff1a;ff_h2645_extract_rbsp函数分析》&#xff09;&#xff0c…

【沐风老师】3DMAX建筑体块生成插件BuildingBlocks使用方法详解

BuildingBlocks建筑体块生成插件使用方法详解 听说你还在手动建配景楼&#xff1f;有了BuildingBlocks这个插件&#xff0c;一分钟搞定喔&#xff01; 3DMAX建筑体块生成插件BuildingBlocks&#xff0c;用于快速自定义街道及生成配景楼区块。 【适用版本】 3dMax2019及更高版…

空间分析在3D应用中的革命:提升投资回报与业务价值

在3D应用的浪潮中&#xff0c;空间分析技术正成为提升用户体验、优化业务决策和解决复杂问题的关键工具。本文将深入探讨空间分析如何通过提供深度用户行为洞察和数据可视化&#xff0c;增强3D应用的实际效益和市场竞争力。 一、空间分析的概念与背景 Tony Bevilacqua&#x…

分布式I/O从站的认知

为什么需要分布式I/O从站&#xff1f; 当PLC与控制机构距离过远时&#xff0c;远距离会带来信号干扰&#xff0c;分布式I/O从站只需要一个网络线缆连接。 ET200分布式I/O从站家族 体积紧凑、功能强大。 ET200SP ET200M ET200S ET200iSP ET200 AL ET200pro ET200 eco PN 通讯协议…

DSSM双塔特征交互

传统的DSSM双塔无法在早期进行user和item侧的特征交互&#xff0c;这在一定程度上降低了模型性能。我们想要对双塔模型进行细粒度的特征交互&#xff0c;同时又不失双塔模型离线建向量索引的解耦性。下面介绍两篇这方面的工作。 美团-Dual Augmented Two-tower 在user和item的特…

1. CSS Grid 网格布局教程

CSS Grid 网格布局教程 一、概述 网格布局&#xff08;Grid&#xff09;是最强大的 CSS 布局方案。 它将网页划分成一个个网格&#xff0c;可以任意组合不同的网格&#xff0c;做出各种各样的布局。以前&#xff0c;只能通过复杂的 CSS 框架达到的效果&#xff0c;现在浏览器…

linux工具应用_VERDI

verdi 1. 基础知识1.1 verdi介绍1.2 fsdb文件2. fsdb dump2.1 1st step-设置环境变量LD_LIBRARY_PATH2.2 2nd step-xrun仿真命令2.3 3rd step-仿真过程中调用fsdb函数dump波形2.3.1 在testbench、top.sv中调用fsdb函数2.3.2 在tcl脚本中用xrun的dump指令(同样要调用fsdb函数)2.…

Scrapy crawl spider 停止工作

Scrapy是一个用于爬取网站数据的流行框架&#xff0c;有时爬虫可能会停止工作&#xff0c;这通常是由多种原因引起的。以下是一些常见问题及其解决方法&#xff1a; 1、问题背景 用户在使用 Scrapy 0.16.2 版本进行网络爬取时遇到问题&#xff0c;具体表现为爬虫在运行一段时间…