Flink实现kafka到kafka、kafka到doris的精准一次消费

1天前 • 大数据

1 流程图

Flink实现kafka到kafka、kafka到doris的精准一次消费

2 Flink来源表建模

--来源-城市topic
CREATE TABLE NJ_QL_JC_SSJC_SOURCE (
record string 
) WITH (
	'connector' = 'kafka',
	'topic' = 'QL_JC_SSJC',
	'properties.bootstrap.servers' = '172.*.*.*:9092',
	'properties.group.id' = 'QL_JC_SSJC_NJ_QL_JC_SSJC_SOURCE',
    'scan.startup.mode' = 'group-offsets',
    'properties.isolation.level' = 'read_committed',
    'properties.auto.offset.reset' = 'earliest',
	'format' = 'raw'
);
--来源-中台kafka-topic
CREATE TABLE ODS_QL_JC_SSJC_SOURCE (
sscsdm string,
extract_time TIMESTAMP,
record string
) WITH (
	'connector' = 'kafka',
	'topic' = 'ODS_QL_JC_SSJC',
	'properties.bootstrap.servers' = '172.*.*.*:21007,172.*.*.*:21007,172.*.*.*:21007',
	'properties.security.protocol' = 'SASL_PLAINTEXT',
	'properties.sasl.kerberos.service.name' = 'kafka',
	'properties.kerberos.domain.name' = 'hadoop.hadoop.com',
	'properties.group.id' = 'ODS_QL_JC_SSJC_SOURCE_ODS_QL_JC_SSJC_SOURCE',
	'scan.startup.mode' = 'group-offsets',
   'properties.auto.offset.reset' = 'earliest',
   'properties.isolation.level' = 'read_committed',
   'sink.semantic' = 'exactly-once',
	'format' = 'json'
);

3 Flink去向表建模

--去向-中台kafka-topic
CREATE TABLE KAFKA_ODS_QL_JC_SSJC_SINK  (
sscsdm string,
extract_time TIMESTAMP,
record string
) WITH (
	'connector' = 'kafka',
	'topic' = 'ODS_QL_JC_SSJC',
	'properties.bootstrap.servers' = '172.*.*.*:21007,172.*.*.*:21007,172.*.*.*:21007',
	'properties.security.protocol' = 'SASL_PLAINTEXT',
	'properties.sasl.kerberos.service.name' = 'kafka',
	'properties.kerberos.domain.name' = 'hadoop.hadoop.com',
	'format' = 'json', 
   'properties.transaction.timeout.ms' = '900000'
);
--去向-Doris表
CREATE TABLE DORIS_ODS_QL_JC_SSJC_SINK (
	sscsdm STRING,
	extract_time TIMESTAMP,
	record STRING
) WITH (
	'connector' = 'doris',
	'fenodes' = '3.*.*.*:8030,3.*.*.*:8030,3.*.*.*:8030',
	'table.identifier' = 'doris_d.ods_ql_jc_ssjc',
	'username' = 'root',
	'password' = '********',
   'sink.properties.two_phase_commit' = 'true' 
);

4 城市Topic至中台Topic的Flinksql

insert into
  KAFKA_ODS_QL_JC_SSJC_SINK
 SELECT
   '320100' as sscsdm,
   CURRENT_TIMESTAMP as extract_time,
   record
 FROM
   NJ_QL_JC_SSJC_SOURCE
 UNION ALL
SELECT
  '320200' as sscsdm,
  CURRENT_TIMESTAMP as extract_time,
  record
FROM
  WX_QL_JC_SSJC_SOURCE
  .
  .
  .
 UNION ALL
 SELECT
   '320583' as sscsdm,
   CURRENT_TIMESTAMP as extract_time,
   record
 FROM
   KS_QL_JC_SSJC_SOURCE

5 中台Topic至Doris的Flinksql

insert into DORIS_ODS_QL_JC_SSJC_SINK
SELECT
  sscsdm,
  CURRENT_TIMESTAMP as extract_time,
  record
FROM
  ODS_QL_JC_SSJC_SOURCE

本文来自网络，不代表协通编程立场，如若转载，请注明出处：https://net2asp.com/6e8c049a7b.html

flink kafka 大数据

赞 (0)

手把手教你制作手机底部导航栏，领导看完都说好

« 上一篇 1天前

深入浅出Python：从零开始搭建自己的Web服务器

下一篇 » 1天前

数据库

软考数据库详细知识点整理（全）

目录第一章计算机系统基本知识 1.1 计算机系统 1.1.1 计算机硬件组成 1.1.2 中央处理单元 1.1.3 数据表示 1.1.4 校验码 1.2 计算机体系结构 1.2…

1天前
前端

Lua语法基本使用

🍓 简介：java系列技术分享(👉持续更新中…🔥) 🍓 初衷:一起学习、一起进步、坚持不懈 🍓 如果…

1天前
前端

h5键盘弹起遮挡输入框的处理

一、前言：混合开发中遇到一个问题，有些型号的安卓机和ios机型，输入框唤起键盘后，输入框会被键盘遮挡，需要手动滑动才能漏出来，影响用户体验二、解决办法: 1.ios和andro…

1天前
前端

15套前端经典实战项目大合集，小白练手必备实战项目

15套前端经典实战项目大合集，悄悄练习，你会惊艳所有人。今日我以内卷为荣，明日内卷以我为荣，不管学习哪门语言都要做出实际的东西来，这个实际的东西就是项目。这里整理了15前端经典…

1天前
前端

idea中yml文件没有提示解决办法

两步解决yml文件不显示提示，yaml文件显示提示问题 1、在插件中心，先下载下图两个插件 2、在Editor=》File Types新增文件类型，文件名匹配规则需要将yaml和y…

1天前
[lighttpd]lighttpd配置http强制跳转https

参考链接 HowToRedirectHttpToHttps – Lighttpd – lighty labs lighttpd版本 / # lighttpd…

前端 1天前
【开发工具】从eclipse到idea的过度

背景随着eclipse相比以前性能慢了不少，idea在开发工具领域越战越猛，市场份额也逐年增加，其体验得了软件工程师的热爱。概要本文只是做了一个简要的记录，简单描述下本人从e…

前端 1天前
前端

橘子学ES实战操作01之集群模式如何实现快照备份

我们知道ES中通过副本在一定意义上实现了数据的备份和高可用。但是我们说万一副本数据丢失了，不小心被rm -f了，你就说逆天不逆天吧，此时要实现数据真正意义上的备份就要使用到快照机制…

1天前
前端

Java-IDEA 启动Tomcat控制台中文乱码解决

1、中文乱码原因基本上大家安装的windows系统本地语言都是选择中文（不会有人选择英文吧？不会吧？不会吧？），也就是GBK编码；而IDEA或者Tomcat日志使用的是UTF…

1天前
ES如何搜索两个索引

在Elasticsearch（ES）中，索引是存储和查询数据的地方，而联合查询是查询多个索引的能力。当你在ES中查询时，可以指定要查询的索引。如果你要查询两个不同的索引，你可以在…

前端 1天前