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

flinkcdc模式去读holo表的binlog的时候报错，怎么解决？

flinkcdc模式去读holo表的binlog的时候报错，怎么解决？Realtime ingestion service (holohub) not enabled or its endpoint invalid. org.apache.flink.table.api.ValidationException: SQL validation failed. Realtime ingestion service (holohub) not enabled or its endpoint invalid.
at org.apache.flink.table.sqlserver.utils.FormatValidatorExceptionUtils.newValidationException(FormatValidatorExceptionUtils.java:41)
at org.apache.flink.table.sqlserver.utils.ErrorConverter.formatException(ErrorConverter.java:125)
at org.apache.flink.table.sqlserver.utils.ErrorConverter.toErrorDetail(ErrorConverter.java:60)
at org.apache.flink.table.sqlserver.utils.ErrorConverter.toGrpcException(ErrorConverter.java:54)
at org.apache.flink.table.sqlserver.FlinkSqlServiceImpl.validateAndGeneratePlan(FlinkSqlServiceImpl.java:1077)
at org.apache.flink.table.sqlserver.proto.FlinkSqlServiceGrpc$MethodHandlers.invoke(FlinkSqlServiceGrpc.java:3692)
at io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:172)
at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:331)
at io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1HalfClosed.runInContext(ServerImpl.java:820)
at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1147)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:622)
at java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.RuntimeException: Realtime ingestion service (holohub) not enabled or its endpoint invalid.
at com.alibaba.ververica.connectors.hologres.utils.JDBCUtils.getHolohubEndpoint(JDBCUtils.java:206)
at com.alibaba.ververica.connectors.hologres.source.HologresTableSource.getScanRuntimeProvider(HologresTableSource.java:354)
at org.apache.flink.table.planner.connectors.DynamicSourceUtils.validateScanSource(DynamicSourceUtils.java:580)
at org.apache.flink.table.planner.connectors.DynamicSourceUtils.prepareDynamicSource(DynamicSourceUtils.java:246)
at org.apache.flink.table.planner.connectors.DynamicSourceUtils.convertSourceToRel(DynamicSourceUtils.java:176)
at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.toRel(CatalogSourceTable.java:118)
at org.apache.calcite.sql2rel.SqlToRelConverter.toRel(SqlToRelConverter.java:3619)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertIdentifier(SqlToRelConverter.java:2523)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2156)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2105)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2062)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:666)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:647)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3472)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:573)
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.java:336)
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.java:329)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.toQueryOperation(SqlToOperationConverter.java:1398)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlQuery(SqlToOperationConverter.java:1326)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertValidatedSqlNode(SqlToOperationConverter.java:384)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:274)
at org.apache.flink.table.planner.delegation.OperationIterator.convertNext(OperationIterator.java:102)
at org.apache.flink.table.planner.delegation.OperationIterator.next(OperationIterator.java:86)
at org.apache.flink.table.planner.delegation.OperationIterator.next(OperationIterator.java:49)
at org.apache.flink.table.sqlserver.execution.OperationExecutorImpl.validate(OperationExecutorImpl.java:396)
at org.apache.flink.table.sqlserver.execution.OperationExecutorImpl.validateAndGeneratePlan(OperationExecutorImpl.java:357)
at org.apache.flink.table.sqlserver.execution.DelegateOperationExecutor.lambda$validateAndGeneratePlan$29(DelegateOperationExecutor.java:263)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1729)
at org.apache.flink.table.sqlserver.context.SqlServerSecurityContext.runSecured(SqlServerSecurityContext.java:72)
at org.apache.flink.table.sqlserver.execution.DelegateOperationExecutor.wrapClassLoader(DelegateOperationExecutor.java:311)
at org.apache.flink.table.sqlserver.execution.DelegateOperationExecutor.lambda$wrapExecutor$35(DelegateOperationExecutor.java:333)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
... 3 more

展开
收起
三分钟热度的鱼 2024-05-16 08:34:58 93 0
1 条回答
写回答
取消 提交回答
  • 圆不溜秋的小猫猫

    使用这个方式创建临时表，就没问题了。CREATE TEMPORARY TABLE IF NOT EXISTS source_student(
    id int,
    。。。
    PRIMARY KEY(id) NOT ENFORCED
    )WITH(
    'connector' = 'hologres',
    'dbname' = 'holo_test', --Hologres的数据库名称
    'tablename' = 'public.source_student', --Hologres用于接收数据的表名称
    'username' = '', --当前阿里云账号的AccessKey ID
    'password' = '    ', --当前阿里云账号的AccessKey Secret
    'endpoint' = 'hgpostcn-**ncs.com:80', --当前Hologres实例VPC网络的Endpoint
    'binlogStartUpMode' = 'initial', -- 先读取历史全量数据，再增量消费Binlog。
    'binlog' = 'true', -- 开启开启实时同步
    'cdcMode' = 'true', -- 开启开启实时同步
    'binlogMaxRetryTimes' = '10', -- 开启开启实时同步
    'binlogRetryIntervalMs' = '500', -- 开启开启实时同步
    'binlogBatchReadSize' = '100', -- 开启开启实时同步
    'jdbcretrycount' = '1', --连接故障时的重试次数
    'partitionrouter' = 'true', --是否写入分区表
    'createparttable' = 'true', --是否自动创建分区
    'mutatetype' = 'insertorignore' --数据写入模式
    );
    -- 设置表属性开启Binlog功能
    begin;
    call set_table_property('source_student', 'binlog.level', 'replica');
    commit;

    -- 设置表属性，配置Binlog TTL时间，单位秒
    begin;
    call set_table_property('source_student', 'binlog.ttl', '2592000');
    commit; 此回答整理自钉群“实时计算Flink产品交流群”

    2024-05-16 09:16:17
    赞同 4 展开评论 打赏
问答分类：
SQL 实时计算 Flink版 云数据库 RDS SQL Server 版 实时数仓 Hologres
问答标签：
flinkcdc binlog binlog模式 holo binlog binlog表 binlog flinkcdc
问答地址：
开发者社区 > 大数据与机器学习 > 实时计算 Flink > 问答

版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

相关问答
实时计算 Flink版
flinkcdc支持任务起动后不重启的情况下mysql source能捕获新建表的binlog吗?
59
1
0
为什么使用FlinkCDC和Kakfa Sink向Kakfa中发送binlog数据，会存在脏数据？
65
0
0
canal -adapter只能适配mysql库binlog_format=row模式？
49
1
0
FLinkcdc 在初始化快照阶段的时候是不是就会记录下binlog点位？
41
1
0
Flink的holo表的binlog最长生命周期是多长啊？
39
1
0
Flink的datastream api消费holo表的binlog的时候起始时间是怎么设置的？
39
1
0
Flink多个表的同步顺序 是按照binlog顺序执行的么？有个表的同步失败了会影响后续执行吗？
31
1
0
云数据仓库ADB打开表的binlog特性，对表会有什么影响吗？
51
1
0
flinkcdc 采集mysql出现卡在某个binlog文件出不来怎么办？
45
0
0
flink-cdc 在抽取mysql binlog，运行一段时间 source 就报java.uti
1039
3
0
大数据与机器学习

实时计算 Flink

实时计算Flink版是阿里云提供的全托管Serverless Flink云服务，基于 Apache Flink 构建的企业级、高性能实时大数据处理系统。提供全托管版 Flink 集群和引擎，提高作业开发运维效率。

我要提问

热门讨论

热门文章

  • FFA 2024 大会门票免费送！AI时代下大数据技术未来路在何方？
    1690
  • 请问Flink 写数据到 ES 有遇到过这个报错的么？
    735
  • flink1.20.0 部署后发布报错，是怎么回事，各种配置都配置了
    297
  • 在建行，数据通常来源于哪些？
    541
  • FlinkCDC MySQL 中 scan.startup.mode 用的是什么模式啊？
    3618
  • Flink如何配置Task Manager？
    705
  • Caused by: org.apache.kafka.common.errors.TimeoutE
    3345
  • 请教下，finksql 这个statettl 哪里设置
    1613
  • 请问flink-connector-jdbc在sink时，如何设置批量写入？
    3183
  • 请问如何用flink sql客户端用yarn application模式提交任务呢？
    1569
    • 展开全部
  • Flink SQL 功能解密系列 —— 流计算“撤回(Retraction)”案例分析
    18479
  • 周末直播|Flink、Hologres、AI等热门话题全都安排！
    4924
  • 直播预告｜FFA 2024 上海主论坛，11月29日正式上线！
    134
  • Flink CDC 3.0 正式发布，详细解读新一代实时数据集成框架
    105184
  • 实时计算 Flash – 兼容 Flink 的新一代向量化流计算引擎
    803
  • 福利「Flink Forward Asia 2023 」视频合集！
    5464
  • 十大行业经典案例！Apache Flink 的 40 个最佳实践
    15814
  • 实时计算引擎 Flink：从入门到深入理解
    1767
  • 实时计算Flink版体验评测
    113
  • Apache Paimon统一大数据湖存储底座
    14299
    • 展开全部

    相关文章

  • 解放数据科学家的神器
  • 网络安全与信息安全：漏洞、加密与安全意识
  • Quick 引擎-抽取性能提升
  • 达梦数据库字段类型 varchar 转 text
  • Flask 框架防止 SQL 注入攻击的方法

    • 相关电子书

    更多
    • PolarDB-X 2.0 全局 Binlog 与备份恢复能 立即下载
    低代码开发师（初级）实战教程 立即下载
    阿里巴巴DevOps 最佳实践手册 立即下载