AI 助理
备案控制台
开发者社区 大数据与机器学习 实时计算 Flink 文章 正文

Flink报错问题之cdc代码报错如何解决

2024-02-25 69
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
实时计算 Flink 版，5000CU*H 3个月
推荐场景：
实时发现最热Github项目
简介： Apache Flink是由Apache软件基金会开发的开源流处理框架，其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。

问题一：flink 1.11.2 cep rocksdb 性能调优

我这边用flink1.11.2 cep做一些模式匹配，发现一旦开启rocksdb做为状态后端，就会出现反压。cpu使用率是之前的10倍。

private void bufferEvent(IN event, long currentTime) throws Exception {

long currentTs = System.currentTimeMillis();

List elementsForTimestamp = elementQueueState.get(currentTime);

if (elementsForTimestamp == null) {

this.bufferEventGetNullhistogram.update(System.currentTimeMillis()

  • currentTs);

elementsForTimestamp = new ArrayList<>();

}else {

this.bufferEventGethistogram.update(System.currentTimeMillis()-currentTs);

}

elementsForTimestamp.add(event);

long secondCurrentTs = System.currentTimeMillis();

elementQueueState.put(currentTime, elementsForTimestamp);

this.bufferEventPuthistogram.update(System.currentTimeMillis() -

secondCurrentTs);

this.bufferEventhistogram.update(System.currentTimeMillis() - currentTs);

}

通过复写CepOperator,加入了一些metics发现

this.bufferEventhistogram = metrics.histogram("buffer_event_delay", new

DescriptiveStatisticsHistogram(1000));

this.bufferEventGethistogram = metrics.histogram("buffer_event_get_delay", new

DescriptiveStatisticsHistogram(1000));

this.bufferEventGetNullhistogram =

metrics.histogram("buffer_event_get_null_delay", new

DescriptiveStatisticsHistogram(1000));

this.bufferEventPuthistogram = metrics.histogram("buffer_event_put_delay", new

DescriptiveStatisticsHistogram(1000));

在get和put比较耗时，整个bufferEvent 能达到200ms

从rocksdb的metric来看没有进行太多flush和compaction。

[image: image.png]

[image: image.png]

也参考了https://www.jianshu.com/p/2e61c2c83c57这篇文章调优过，发现效果也不是很好，一样反压。

也看过类似的问题http://apache-flink.147419.n8.nabble.com/rocksDB-td1785.html

，但是我这sst文件很小。

请教大家，为啥get和put这么耗时呢？有什么好的优化方案不？谢谢。*来自志愿者整理的flink邮件归档



参考答案：

‘@Override

public UV get(UK userKey) throws IOException, RocksDBException {

byte[] rawKeyBytes =

serializeCurrentKeyWithGroupAndNamespacePlusUserKey(userKey,

userKeySerializer);

byte[] rawValueBytes = backend.db.get(columnFamily, rawKeyBytes);

return (rawValueBytes == null ? null :

deserializeUserValue(dataInputView, rawValueBytes,

userValueSerializer));

}

@Override

public void put(UK userKey, UV userValue) throws IOException, RocksDBException {

byte[] rawKeyBytes =

serializeCurrentKeyWithGroupAndNamespacePlusUserKey(userKey,

userKeySerializer);

byte[] rawValueBytes = serializeValueNullSensitive(userValue,

userValueSerializer);

backend.db.put(columnFamily, writeOptions, rawKeyBytes, rawValueBytes);

}

通过源码跟踪发现，RocksDBMapState每次get和put都需要序列化和反序列化。。。应该是这个原因导致比较耗时。*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/371646?spm=a2c6h.13066369.question.84.6ad263826CoWoh



问题二：cdc代码报错

我运行ververica/flink-cdc-connectors git上的demo代码，报错：

2020-11-04 16:39:10.972 [Source: Custom Source -> Sink: Print to Std. Out (1/1)] WARN org.apache.flink.runtime.taskmanager.Task - Source: Custom Source -> Sink: Print to Std. Out (1/1) (7c3ccf7686ccfb33254e8cb785cd339d) switched from RUNNING to FAILED.

java.lang.AbstractMethodError: org.apache.kafka.connect.json.JsonSerializer.configure(Ljava/util/Map;Z)V

at org.apache.kafka.connect.json.JsonConverter.configure(JsonConverter.java:300)

at org.apache.kafka.connect.json.JsonConverter.configure(JsonConverter.java:311)

at io.debezium.embedded.EmbeddedEngine. (EmbeddedEngine.java:583)

at io.debezium.embedded.EmbeddedEngine. (EmbeddedEngine.java:80)

at io.debezium.embedded.EmbeddedEngine$BuilderImpl.build(EmbeddedEngine.java:301)

at io.debezium.embedded.EmbeddedEngine$BuilderImpl.build(EmbeddedEngine.java:217)

at io.debezium.embedded.ConvertingEngineBuilder.build(ConvertingEngineBuilder.java:139)

at com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction.run(DebeziumSourceFunction.java:299)

at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)

at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)

at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213)

源码：

public class CdcTest {

public static void main(String[] args) throws Exception {

SourceFunction sourceFunction = MySQLSource. builder()

.hostname("localhost")

.port(3306)

.databaseList("sohay") // monitor all tables under inventory database

.username("root")

.password("123456")

.deserializer(new StringDebeziumDeserializationSchema()) // converts SourceRecord to String

.build();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.addSource(sourceFunction).print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

env.execute();

}

}*来自志愿者整理的flink邮件归档



参考答案：

环境中估计有另一个版本的 kafka-connect jar 包，导致依赖冲突了。*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/371645?spm=a2c6h.13066369.question.83.6ad26382ziEiEA



问题三：使用flink-CDC checkpoint超时问题

我刚调研flink sql cdc功能，我有一个这样的需求，就是三张CDC的表做left join关联，由于三张表都会变化，所以使用cdc来做。

前两张表数据每张大概20万条数据，最后一张表只有几十条数据，我是讲三张表关联之后做成宽表写入的mysql中。

每次数据写到10万条左右任务失败，查看了一下日志，是checkpoint超时造成的。状态后端用的rocksDB。我想问

数据量也不是很大，为什么checkpoint会超时呢，也就是10分钟没有完成checkpoint？*来自志愿者整理的flink邮件归档



参考答案：

我建议可以从两方面排查一下：

1、检查 checkpoint 的大小，是不是很大？

2、检查作业是否反压？反压的情况下 checkpoint 一般很难成功，这种情况可以先解决反压的问题。*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/371644?spm=a2c6h.13066369.question.86.6ad263822mbZYT



问题四：flink1.11.0 sql自定义UDAF包含复合类型时报Incompatible types

本人使用flink版本为1.11.0，自定义udaf如下：

public class GetContinuousListenDuration extends AggregateFunction<Row, ContinuousListenDuration> {

private static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm");

@Override

@DataTypeHint("ROW<startTime TIMESTAMP(3), duration BIGINT>")

public Row getValue(ContinuousListenDuration acc) {

return Row.of(acc.getStartTime(), acc.getDuration());

}

@Override

public ContinuousListenDuration createAccumulator() {

return new ContinuousListenDuration();

}

public void accumulate(ContinuousListenDuration acc, @DataTypeHint("TIMESTAMP(3)") LocalDateTime dt, Boolean isListening) {

// 此处省略逻辑

}

}

聚合时以Timestamp(3)、Boolean作为参数，getValue返回类型是ROW<startTime TIMESTAMP(3), duration BIGINT>，函数名定义为get_continuous_listen_duration，调用该函数的sql如下：

insert into

report.result

select

id,

city_code,

get_continuous_listen_duration(

dt,

(order_no is null)

or (trim(order_no) = '')

).startTime as start_time,

get_continuous_listen_duration(

dt,

(order_no is null)

or (trim(order_no) = '')

).duration as duration

from

(

select

o.id,

o.dt,

o.order_no,

r.city_code

from

(

select

req [1] as id,

dt,

proctime,

req [2] as order_no

from

tmp_v

where

extra [1] is null

or extra [1] <> 'false'

) o

JOIN dim.right FOR SYSTEM_TIME AS OF o.proctime AS r ON r.id = o.id

) a

group by

id,

city_code

having

get_continuous_listen_duration(

dt,

(order_no is null)

or (trim(order_no) = '')

).duration >= 2

运行时发生如下异常：

Caused by: org.apache.calcite.sql.validate.SqlValidatorException: Incompatible types

at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[?:1.8.0_171]

at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) ~[?:1.8.0_171]

at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[?:1.8.0_171]

at java.lang.reflect.Constructor.newInstance(Constructor.java:423) ~[?:1.8.0_171]

at org.apache.calcite.runtime.Resources$ExInstWithCause.ex(Resources.java:457) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.runtime.Resources$ExInst.ex(Resources.java:550) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:839) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:824) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.fun.SqlDotOperator.deriveType(SqlDotOperator.java:101) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SqlValidatorImpl$DeriveTypeVisitor.visit(SqlValidatorImpl.java:5858) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SqlValidatorImpl$DeriveTypeVisitor.visit(SqlValidatorImpl.java:5845) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.SqlCall.accept(SqlCall.java:139) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SqlValidatorImpl.deriveTypeImpl(SqlValidatorImpl.java:1800) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SqlValidatorImpl.deriveType(SqlValidatorImpl.java:1785) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.type.InferTypes.lambda$static$0(InferTypes.java:46) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SqlValidatorImpl.inferUnknownTypes(SqlValidatorImpl.java:1980) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SqlValidatorImpl.validateHavingClause(SqlValidatorImpl.java:4214) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SqlValidatorImpl.validateSelect(SqlValidatorImpl.java:3515) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SelectNamespace.validateImpl(SelectNamespace.java:60) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.AbstractNamespace.validate(AbstractNamespace.java:84) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SqlValidatorImpl.validateNamespace(SqlValidatorImpl.java:1110) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SqlValidatorImpl.validateQuery(SqlValidatorImpl.java:1084) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.SqlSelect.validate(SqlSelect.java:232) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:1059) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:766) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:141) ~[flink-table-blink_2.11-1.11.0.jar:1.11.0]

at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:108) ~[flink-table-blink_2.11-1.11.0.jar:1.11.0]

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:187) ~[flink-table-blink_2.11-1.11.0.jar:1.11.0]

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlInsert(SqlToOperationConverter.java:527) ~[flink-table-blink_2.11-1.11.0.jar:1.11.0]

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:204) ~[flink-table-blink_2.11-1.11.0.jar:1.11.0]

at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:78) ~[flink-table-blink_2.11-1.11.0.jar:1.11.0]

at org.apache.flink.table.api.internal.StatementSetImpl.addInsertSql(StatementSetImpl.java:52) ~[flink-table_2.11-1.11.0.jar:1.11.0]

at com.ververica.flink.table.gateway.operation.MultiSqlOperation.lambda$executeInternal$0(MultiSqlOperation.java:119) ~[flink-sql-gateway-0.1-SNAPSHOT.jar:?]

at com.ververica.flink.table.gateway.context.ExecutionContext.wrapClassLoader(ExecutionContext.java:223) ~[flink-sql-gateway-0.1-SNAPSHOT.jar:?]

at com.ververica.flink.table.gateway.operation.MultiSqlOperation.executeInternal(MultiSqlOperation.java:109) ~[flink-sql-gateway-0.1-SNAPSHOT.jar:?]

我的问题是这样定义udf有什么问题吗？*来自志愿者整理的flink邮件归档



参考答案：

从这行报错堆栈来看：at org.apache.calcite.sql.fun.SqlDotOperator.deriveType(SqlDotOperator.java:101) ， 应该是在对 row.startTime 或者 row. duration validate 阶段，推断类型时识别出不兼容类型，可以检测下用法有没有错误。*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/371643?spm=a2c6h.13066369.question.85.6ad26382Qobcby



问题五：flink 1.11.1 web ui 页面查看source算子的detail数据，recoreds

    我的flink程序正常执行，但是我在web ui监控页面查看source算子的detail信息，里面的Records Sent等度量信息，永远为0。请问是什么问题？*来自志愿者整理的flink邮件归档



参考答案：

你任务的DAG是什么样子的呢，可能的原因：

1.source本来就没有收到数据，或者没有发送到下游

2.source和下游算子chain在一起看不出来*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/371642?spm=a2c6h.13066369.question.88.6ad26382p86CnM

文章标签：
实时计算 Flink版
云消息队列 Kafka 版
流计算
Apache
SQL
Java
监控
关键词：
实时计算 Flink版报错
实时计算 Flink版CDC
实时计算 Flink版cdc代码
实时计算 Flink版代码
实时计算 Flink版报错cdc
相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程，适合初学者阅读。由浅入深案例丰富，通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员，只要能够坚持把所有章节都学完，也一定会受益匪浅。
三分钟热度的鱼
目录
相关文章
灵杰开发者
|
3天前
|
Oracle 关系型数据库 新能源
Flink CDC 在新能源制造业的实践
本文撰写自某新能源企业的研发工程师 单葛尧 老师。本文详细介绍该新能源企业的大数据平台中 CDC 技术架构选型和 Flink CDC 的最佳实践。
灵杰开发者
267 5
Flink CDC 在新能源制造业的实践
三分钟热度的鱼
|
13天前
|
SQL Shell API
实时计算 Flink版操作报错合集之任务提交后出现 "cannot run program "/bin/bash": error=1, 不允许操作" ，是什么原因
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
53 1
实时计算 Flink版操作报错合集之任务提交后出现 "cannot run program "/bin/bash": error=1, 不允许操作" ，是什么原因
三分钟热度的鱼
|
13天前
|
资源调度 监控 关系型数据库
实时计算 Flink版操作报错合集之处理大量Join时报错空指针异常，是什么原因
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
12 1
实时计算 Flink版操作报错合集之处理大量Join时报错空指针异常，是什么原因
防火防爆
|
10天前
|
SQL 数据库 流计算
Flink CDC数据读取问题之一致性如何解决
Flink CDC 使用Change Data Capture (CDC)技术从数据库捕获变更事件，并利用Flink的流处理能力确保数据读取一致性。相较于传统工具，它具备全增量一体化数据集成能力，满足实时性需求。在实践中解决了高效数据同步、稳定同步大量表数据等问题。应用场景包括实时数据同步、实时数据集成等。快速上手需学习基本概念与实践操作。未来发展方向包括提升效率与稳定性，并依据用户需求持续优化。
防火防爆
42 1
三分钟热度的鱼
|
13天前
|
SQL Java Apache
实时计算 Flink版操作报错合集之使用parquet时，怎么解决报错:无法访问到java.uti.Arrays$ArrayList类的私有字段
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
12 1
三分钟热度的鱼
|
13天前
|
Oracle 关系型数据库 Java
实时计算 Flink版操作报错合集之遇到了关于MySqIValidator类缺失的错误，是什么原因
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
10 1
三分钟热度的鱼
|
13天前
|
SQL 存储 资源调度
实时计算 Flink版操作报错合集之启动项目时报错缺少MySqlValidator类，是什么原因
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
13 1
三分钟热度的鱼
|
13天前
|
Java 关系型数据库 MySQL
实时计算 Flink版操作报错合集之在使用批处理模式中使用flat_aggregate函数时报错，该如何解决
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
12 1
三分钟热度的鱼
|
13天前
|
SQL Oracle NoSQL
实时计算 Flink版操作报错合集之报错“找不到对应的归档日志文件”，怎么处理
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
26 1
灵杰开发者
|
2月前
|
存储 监控 大数据
阿里云实时计算Flink在多行业的应用和实践
本文整理自 Flink Forward Asia 2023 中闭门会的分享。主要分享实时计算在各行业的应用实践，对回归实时计算的重点场景进行介绍以及企业如何使用实时计算技术，并且提供一些在技术架构上的参考建议。
灵杰开发者
641 7
阿里云实时计算Flink在多行业的应用和实践
大数据与机器学习

实时计算 Flink

热门文章

最新文章

  • 1
    Apache Paimon V0.9最新进展
  • 2
    阿里云实时计算产品案例&解决方案汇总
  • 3
    Flink Checkpoint 问题排查实用指南
  • 4
    Flink SQL 功能解密系列 —— 数据去重的技巧和思考
  • 5
    使用 Kafka 和 Flink 构建实时数据处理系统
  • 6
    携程实时智能检测平台建设实践
  • 7
    Apache Flink 漫谈系列(06) - 流表对偶(duality)性
  • 8
    Flink 在人工智能领域的应用实践
  • 9
    实时计算无线数据分析
  • 10
    如何分析及处理 Flink 反压？
  • 1
    用实时计算释放当下企业大数据潜能
    115
  • 2
    实时计算 Flink版产品使用问题之如何高效地将各分片存储并跟踪每个分片的消费位置
    31
  • 3
    实时计算 Flink版产品使用问题之使用CTAS同步MySQL到Hologres时出现的时区差异，该如何解决
    7
  • 4
    实时计算 Flink版产品使用问题之两个数据表是否可以同时进行双向的数据同步
    10
  • 5
    实时计算 Flink版产品使用问题之如何将PostgreSQL数据实时入库Hive并实现断点续传
    29
  • 6
    实时计算 Flink版产品使用问题之如何使用Kafka Connector将数据写入到Kafka
    10
  • 7
    实时计算 Flink版产品使用问题之怎么关闭HDFS的Web界面
    9
  • 8
    实时计算 Flink版产品使用问题之如何实现ZooKeeper抖动导致任务失败时，能从最近的检查点重新启动任务
    17
  • 9
    实时计算 Flink版产品使用问题之处理Kafka数据顺序时，怎么确保事件的顺序性
    10
  • 10
    实时计算 Flink版产品使用问题之如何从savepoint重新启动作业
    16

    • 相关产品

  • 实时计算 Flink版
    文档详情 产品详情

    • 相关课程

    更多
  • 实时计算 Flink 版产品入门与实操
  • 开源 Flink 极速上手教程
  • 大数据实时计算框架Spark快速入门
  • Apache Flink 入门

    • 相关电子书

    更多
  • Flink CDC Meetup PPT - 龚中强
  • Flink CDC Meetup PPT - 王赫
  • Flink CDC Meetup PPT - 覃立辉

    • 相关实验场景

    更多
  • 使用Flink实时发现最热Github项目
  • 5分钟上手Flink MySQL连接器
  • 基于Flink+Tair搭建实时监控大屏
  • 基于Hologres+Flink搭建GitHub实时数据大屏
    • 下一篇
    云函数