flinkcdc（flinkcdc支持哪些库）

by intanet.cn ca 大数据 on 2024-03-18

本篇文章给大家谈谈flinkcdc，以及flinkcdc支持哪些库对应的知识点，希望对各位有所帮助，不要忘了收藏本站喔。

本文目录一览：

1、疫情期间是听乡政府的还是疾控中心的？
2、hudi做ETL大数据量情况下聚合数据不正确问题分享
3、flink sql cdc mysql建表报错了???.DataException: NAME is not a valid field name
4、“数据湖三剑客”Hudi、Delta Lake和Iceberg 深度对比
5、如何同步mysql数据到Doris中
6、Flink SQL实战演练之自定义Table Format

疫情期间是听乡政府的还是疾控中心的？

疫情期间，肯定是听政府的啊！疾控也得根据政府的指令做事啊！疫情期间，肯定是要居家隔离的啊！这个具体情况具体对待！

疾病预防控制中心

主管国家疾病预防控制的业务机构

疾病预防控制中心，简称疾控中心，如：四川省疾病预防控制中心；沈阳军区疾控中心。“疾病控制中心”一词来自美国主管国家疾病预防控制的业务机构。

截至2021年末，全国共有疾病预防控制中心3380个。[1]

中文名

疾病预防控制中心

外文名

Center for Disease Control and Prevention

英文简称

CDC

中文简称

疾控中心

hudi做ETL大数据量情况下聚合数据不正确问题分享

1.问题： 17W明细数据批量导入hudi,经过路径mysql-hudi ods表-hudi dws表-mysql 的运算后，发现sink到mysql表的聚合结果有误，详细过程如下简乎：

Case:按天计算报表查询最大耗时指标：

经过以上数据流转后，mysql结果表如右图，无线外购报表的最大耗时计算错误

2.使用软件版本：

hudi:0.10.0;

flink: flink-1.13.1

flink-cdc:2.0

3.原因分析：

step1:经过测试，发现只有在大并发数据插入情况下，会出现数据计算错误问题，且出问题的数据比较随机，小数据量情况下的增删改均无问题，考虑是并发场景下数据流转方面出现问题；

step2:查询Hudi-DWS表，发现聚合结果是正确的，排除了Hudi聚合计算流程问题，将目标定位到从Hudi-DWS表 sink 数据到mysql结果表任务，该任务对应的SQL为：

INSERT INTO item_access_groupby_jdbc SELECT `item_name` ,`group_name` ,`report_name` ,`report_id`,`avg_delay`,`max_delay`,`min_delay` ,`access_type` ,`createdate` ,CAST(`access_count` AS int) ,CAST(`access_user_count` AS int) FROM dws_item_access_groupby_hudi;

任务Job对应的执行图如下：

通过Job图和源码分析，整个读取Hudi-DWS表，并Sink到mysql结果表的过程分为如下几步：

(1)split_monitor:每间隔3s(可设置)去监听Hudi TimeLine 上是否存在新提交的Instance，有的话则读取新提交的Instance，并获取对应的数据FileSlice(内包含数据log文件和parquet文件)，将信息封装，下发给后面的split_reader进行处理，这里要注意其分发模式是Rebanlance；

(2)split_reader：接收split_monitor传递的需要处理的文件信息，对文件中的数据进行处理，完毕后sink到mysql中，注意这里的并发度为4；

所以猜测split_monitor在两次监控下，对同一数据文件进行了Rebanlance模式的分发，并分发给了两个不同的task进行处理，task对数据文件处理速度不一致，导致了老的回撤流覆盖了新的回撤流，流程如下：

数据文件结构：

数据导出流程：

4.解决方法：

方案一：将split_reader并行度指定为1，此时只有一个task处理log数据文件，保证处理顺序性，具体改动是在定义Hudi-DWS表的时候指定参数'read.tasks' = '1'，但该方案会影响sink处理速度；

方案二：修改源码汪氏：在分发log文件时候，按照分区值进行keyBy，保证同一分区下数据文件都给一个Task进行处理，从而保证数据处理的有序性，主要修改如下三个类：

1.HoodieTableSource类：

2.IncrementalInputSplits类的inputSplits方法：

3.MergeOnReadInputSplit类，增加变量realPartition(表示分困咐散区)

修改完成后，重新进行hudi打包，问题消失，mysql结果表数据计算正确，job图如下，log分发从之前的Rebanlance变成了hash，此时同一个log文件被相同task处理，保障了处理的顺序性。

flink sql cdc mysql建表报错了???.DataException: NAME is not a valid field name

你好，很高兴回答你的问题。

建议检查一下是不是有“NAME”这个字段。滑斗樱

如果有帮助到你，销老请点击采纳信丛。

[img]

“数据湖三剑客”Hudi、Delta Lake和Iceberg 深度对比

一个热爱生活又放荡不羁的程序猿

本文主要讲解如下内容：

一、数据湖的优点

二、目前有哪些开源数据湖组件

三、三大数据湖组件对比

数据湖相比传统数仓而言，最明显的便是优秀的T+0能力，这个解决了Hadoop时代数据分析的顽疾。传统的数据处理流程从数据入库到数据处理通常需要一个较长的环节、涉及许多复杂的逻辑来保证数据的一致性，由于架构的复杂性使得整个流水线具有明显的延迟。

目前开源的数据湖有江湖人称“数据湖三剑客”的 Hudi、Delta Lake和Iceberg

Iceberg官网定义：Iceberg是一个通用的表格式（数据组织格式），提供高性能的读写和元数据管理功能。

Iceberg 的 ACID 能力可以简化整个流水线的设计，传统 Hive/Spark 在修正数据时需要将数据读取出来，修改后再写入，有极大的修正成本。

[玫瑰]ACID能力，无缝贴合流批一体数据存储

随着flink等技术的不断发展，流批一体生态不断完善，但在流批一体数据存储方面一直是个空白，直到Iceberg等数据湖技术的出现，这片空白被慢慢填补。

Iceberg 提供 ACID 事务能力，上游数据写入即可见，不影响当前数据处理任务，这大大简化了 ETL；

Iceberg 提供了 upsert、merge into 能力，可以极大地缩小数据入库延迟；

[玫瑰]统一数据存储，无缝衔接计算引擎和数据存储

Iceberg提供了基于流式的增量计算模型和基于批处理的全量表计算模型。批处理和流任务可以使用相同的存储模型，数据不再孤立；

Iceberg 支持隐藏分区和分区进化，方便业务进行数据分区策略更新。

Iceberg屏蔽了底层数据存枣罩储格式的差异，提供对于Parquet，ORC和Avro格式的支持。将上层引擎的能力传导到下层的存储格式。

[玫瑰]开放架构设计，开发维护成本相对可控

Iceberg 的架构和实现并未绑定于某一特定引擎，它实现了通用的数据组织格式，利用此格式可以方便地与不同引擎对接，目前 Iceberg 支持的计算引擎有 Spark、Flink、Presto 以及 Hive。

相比于 Hudi、Delta Lake，Iceberg 的架构实现更为优雅，同时对于数据格式、类型系统有完备的定义和可进化的设计；

面向对象存储的优化。Iceberg 在数据组织方式上充分考虑了对象存储的特性，避免耗时的 listing 和 rename 操作，使其在基于对象存储的数据湖架构适配上更有优势。

[玫瑰]增量数据读取，实时计算的一把利剑

Iceberg 支持通过流式方式读取增量数据，支持 Structed Streaming 以及 Flink table Source。

Apache Hudi是一种数据湖的存储格式，在Hadoop文件系统之上提供了更新数据和删除数据的能力以及消费变化数据的能力。

Hudi支持如下两种表类型：

使用Parquet格式存储数据。Copy On Write表的更新操作需要通过重写实现。

使用列式文件格式（Parquet）和行式文件格式（Avro）混合的方式来存储数据。Merge On Read使用列式格式存放Base数据，同时使用行式格式存放增量数据。最新写入的增量数据存放至行式文件中，根据可配置的策略执行COMPACTION操作合并增量数据至列式文件中。

应用场景

Hudi支持插入、更新和删除数据。可以实时消费消息队列（Kafka）和日志服务SLS等日志数据至Hudi中，同时也支持实时同步数游岩梁据库Binlog产生的变更数据。

Hudi优化了数据写入过程中产生的小文件。因此，相比其神运他传统的文件格式，Hudi对HDFS文件系统更加的友好。

Hudi支持多种数据分析引擎，包括Hive、Spark、Presto和Impala。Hudi作为一种文件格式，不需要依赖额外的服务进程，在使用上也更加的轻量化。

Hudi支持Incremental Query查询类型，可以通过Spark Streaming查询给定COMMIT后发生变更的数据。Hudi提供了一种消费HDFS变化数据的能力，可以用来优化现有的系统架构。

Delta Lake是Spark计算框架和存储系统之间带有Schema信息数据的存储中间层。它给Spark带来了三个最主要的功能：

第一，Delta Lake使得Spark能支持数据更新和删除功能；

第二，Delta Lake使得Spark能支持事务；

第三，支持数据版本管理，运行用户查询历史数据快照。

核心特性

Delta lake

由于Apache Spark在商业化上取得巨成功，所以由其背后商业公司Databricks推出的Delta lake也显得格外亮眼。在没有delta数据湖之前，Databricks的客户般会采经典的lambda架构来构建他们的流批处理场景。

Hudi

Apache Hudi是由Uber的程师为满其内部数据分析的需求设计的数据湖项，它提供的fast upsert/delete以及compaction等功能可以说是精准命中民群众的痛点，加上项各成员积极地社区建设，包括技术细节分享、国内社区推等等，也在逐步地吸引潜在户的光。

Iceberg

Netflix的数据湖原先是借助Hive来构建，但发现Hive在设计上的诸多缺陷之后，开始转为研Iceberg，并最终演化成Apache下个度抽象通的开源数据湖案。

三者均为Data Lake的数据存储中间层，其数据管理的功能均是基于系列的meta 件。Meta 件的类似于数据库的catalog，起到schema管理、事务管理和数据管理的功能。与数据库不同的是，这些meta 件是与数据件起存放在存储引擎中的，户可以直接看到。这个做法直接继承了数据分析中数据对户可见的传统，但是形中也增加了数据被不破坏的风险。旦删了meta 录，表就被破坏了，恢复难度很。

Meta包含有表的schema信息。因此系统可以掌握schema的变动，提供schema演化的持。Meta 件也有transaction log的功能（需要件系统有原性和致性的持）。所有对表的变更都会成份新的meta 件，于是系统就有了ACID和多版本的持，同时可以提供访问历史的功能。在这些，三者是相同的。

Hudi 的设计标正如其名，Hadoop Upserts Deletes and Incrementals（原为 Hadoop Upserts anD Incrementals），强调了其主要持Upserts、Deletes 和 Incremental 数据处理，其主要提供的写具是 Spark HudiDataSource API 和提供的 HoodieDeltaStreamer，均持三种数据写式：UPSERT，INSERT 和 BULK_INSERT。其对 Delete 的持也是通过写时指定定的选项持的，并不持纯粹的 delete 接。

在查询，Hudi 持 Hive、Spark、Presto。

在性能，Hudi 设计了 HoodieKey ，个类似于主键的东西。对于查询性能，般需求是根据查询谓词成过滤条件下推 datasource。Hudi 这没怎么做作，其性能完全基于引擎带的谓词下推和 partition prune 功能。

Hudi 的另特是持 Copy On Write 和 Merge On Read。前者在写时做数据的 merge，写性能略差，但是读性能更些。后者读的时候做 merge，读性能差，但是写数据会较及时，因后者可以提供近实时的数据分析能。最后，Hudi 提供了个名为run_sync_tool 的脚本同步数据的 schema 到 Hive 表。Hudi 还提供了个命令具于管理 Hudi 表。

Iceberg 没有类似的 HoodieKey 设计，其不强调主键。没有主键，做 update/delete/merge 等操作就要通过 Join 来实现， Join 需要有个类似 SQL 的执引擎。

Iceberg 在查询性能做了量的作。值得提的是它的 hidden partition 功能。Hidden partition 意思是说，对于户输的数据，户可以选取其中某些列做适当的变换（Transform）形成个新的列作为 partition 列。这个 partition 列仅仅为了将数据进分区，并不直接体现在表的 schema中。

Delta 的定位是流批体的，持 update/delete/merge，spark 的所有数据写式，包括基于dataframe 的批式、流式，以及 SQL 的 Insert、Insert Overwrite 等都是持的。

不强调主键，因此其 update/delete/merge 的实现均是基于 spark 的 join 功能。在数据写，Delta 与 Spark 是强绑定的，这点 Hudi 是不同的：Hudi 的数据写不绑定 Spark。

在查询，Delta 前持 Spark 与 Presto，但是，Spark 是不可或缺的，因为 delta log 的处理需要到 Spark。这意味着如果要 Presto 查询 Delta，查询时还要跑个 Spark 作业。更为难受的是，Presto 查询是基于 SymlinkTextInputFormat 。在查询之前，要运 Spark 作业成这么个 Symlink 件。如果表数据是实时更新的，意味着每次在查询之前先要跑个 SparkSQL，再跑 Presto。为此，EMR 在这做了改进可以不必事先启动个 Spark 任务。

在查询性能，开源的 Delta 乎没有任何优化。

Delta 在数据 merge 性能不如 Hudi，在查询性能不如 Iceberg，是不是意味着 Delta 是处了呢？其实不然。Delta 的优点就是与 Spark 的整合能，尤其是其流批体的设计，配合 multi-hop 的 data pipeline，可以持分析、Machine learning、CDC 等多种场景。使灵活、场景持完善是它相 Hudi 和 Iceberg 的最优点。另外，Delta 号称是 Lambda 架构、Kappa 架构的改进版，需关流批，需关架构。这点上 Hudi 和 Iceberg 是所不及的。

三个引擎的初衷场景并不完全相同，Hudi 为了 incremental 的 upserts，Iceberg 定位于性能的分析与可靠的数据管理，Delta 定位于流批体的数据处理。这种场景的不同也造成了三者在设计上的差别。尤其是 Hudi，其设计与另外两个相差别更为明显。

Delta、Hudi、Iceberg三个开源项中，Delta和Hudi跟Spark的代码深度绑定，尤其是写路径。这两个项设计之初，都基本上把Spark作为他们的默认计算引擎了。 Apache Iceberg的向常坚定，宗旨就是要做个通化设计的Table Format。

Iceberg完美的解耦了计算引擎和底下的存储系统，便于多样化计算引擎和件格式，很好的完成了数据湖架构中的Table Format这层的实现，因此也更容易成为Table Format层的开源事实标准。另，Apache Iceberg也在朝着流批体的数据存储层发展，manifest和snapshot的设计，有效地隔离不同transaction的变更，常便批处理和增量计算。并且，Apache Flink已经是个流批体的计算引擎，者都可以完美匹配，合打造流批体的数据湖架构。

Apache Iceberg这个项背后的社区资源常丰富。在国外，Netflix、Apple、Linkedin、Adobe等公司都有PB级别的产数据运在Apache Iceberg上；在国内，腾讯这样的巨头也有常庞的数据跑在Apache Iceberg之上，最的业务每天有 T的增量数据写。

如何同步mysql数据到Doris中

Doris官网定义

mysql原始表结构

1.doris中关联mysql外表

结果如下：

2.doris中关联kafka导入数据

查看作业

State为RUNNING，表示已经成功。

停止作业

3.通过flink导入mysql数据到doris

方法1：通过mysql-cdc写入kafka，kafka关联doris表。

方法2：通过阿里云DTS-datahub，然后通过Flink写入kafka，再关联到doris外表

如何处理delete数据？对于方法1，需要手动的删除doris中的数据；对于方法2，可以通过dts_operation_flag字段来标模败瞎示，dts_operation_flag可以为I/U/D，分别表示添加、更新和删除。那我们就只需要在doris表中添加一个dts_operation_flag字段来标示就可以了，查询数据的时候就不再查询等于D的值。

如何处理脏数据？delete doris中的数据，然后insert正确的值；还有个旦空方法是将关联一个外表（这个是正确的值），然后再枯慧将doris中的表和外表中的值diff，将diff的值insert到doris中。

Flink SQL实战演练之自定义Table Format

简介：接着上次Flink CDC继续带歼聊，提到这块，不得不重点说下canal-json format了，canal json format对json format进行了封装，负责把binlog形式的json数据转化成了Flink能够识别的RowData数据，当然也包括数据的描述信息封装类RowType。笔者想实现根据RowKind进行数据的过滤，目前可以通过修改canal format的源数据来实现，也可以通过将changelog流以changelog json的形式回写Kafka。

基于目前对纳谈table format的了解，这里自定义event json format，用来处理事件流数据，因为事件流字段不固定，可能只有少部分字段是固洞行碰定的，其他字段都是扩展的，所以笔者想实现用户自定义schema指定公共字段，然后其他字段以json的行为存在metadata中的default字段中。

Table Format作为Connector组件单独用于序列化和反序列化内部数据的模块而单独存在，多个Connector可以公用。自定义Table Format可以让大家更好的理解Flink SQL时如何将外部数据转化为内部可以识别的RowData数据结构的，从而在排查问题的时候能准确定位到具体位置。

关于flinkcdc和flinkcdc支持哪些库的介绍到此就结束了，不知道你从中找到你需要的信息了吗？如果你还想了解更多这方面的信息，记得收藏关注本站。

免费安卓虚拟机app（虚拟安卓手机系统） hive和hadoop的关系（hive和hadoop关系?）