备案控制台
开发者社区 > 大数据与机器学习 > 实时计算 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 24 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 flinkcdc binlog表 binlog报错 holo binlog
问答地址：
开发者社区 > 大数据与机器学习 > 实时计算 Flink > 问答

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

相关问答
实时计算 Flink版
flink报错The binlog event doesn't contain expecte...
17
1
0
flinkcdc在监听binlog的时候如果程序停掉了，如何能让程序在停掉的那个点重新监听呢？
13
1
0
在Flink CDC中重启任务时报了这个错，有一个表的binlog日志不存在，如何还能让任务正常跑？
36
1
0
Hologres中的同一张表的binlog是否是递增的
6
1
0
实时计算Flink报错binlog probably contains events gene...
6
1
0
mysqlbinlog工具导出PolarDB binlog日志报错数据库版本不匹配
5
1
0
Hologres表开启Hologres binlog 后建表时Binlog的TTL设置比较小的...
10
1
0
Flink报错Cannot read the binlog filename and pos...
13
1
0
Hologres用Flink或者holo client往binlog表里一次写入多条重复的数据...
9
1
0
Flink的ODPS离线表的binlog产生的tps有多大？
29
0
0
大数据与机器学习

实时计算 Flink

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

我要提问

热门讨论

热门文章

  • flink1.15启动后无法访问webui的问题有人遇到过吗
    10100
  • FlinkCDC MySQL 中 scan.startup.mode 用的是什么模式啊？
    2120
  • Flink CDC有遇见这个问题的吗？
    728
  • 不同版本的 flink 如何适配不同版本 kafka，flink-connect-kafka 版本
    241
  • Flink CDC中我使用了自定义聚合函数，但是报错这个需要怎么解决呢？
    588
  • Caused by: org.apache.kafka.common.errors.TimeoutE
    2471
  • 使用Flink CDC，生产环境要申请的服务器如何评估
    959
  • Flink sql将数组炸开，实现hive的explode函数的效果，还有什么其他好的方式？
    613
  • flink sql 批模式需要咋配置啊？
    899
  • flink web UI 是不是本身不支持登录认证，必须通过nginx来加登录认证？
    777
    • 展开全部
  • Flink 流批一体场景应用及落地情况
    67302
  • 湖仓一体全面开启实时化时代
    50545
  • 数据仓库介绍与实时数仓案例
    41390
  • 独家专访阿里集团副总裁贾扬清：我为什么选择加入阿里巴巴？
    17439
  • Flume+Kafka+Flink+Redis构建大数据实时处理系统：实时统计网站PV、UV展示
    22436
  • 阿里云实时计算产品案例&解决方案汇总
    26433
  • 回顾 | Kafka x Flink Meetup 与世界人工智能大会大数据 AI 专场精彩回顾（附PPT下载）
    13917
  • 为什么说流处理即未来？
    9897
  • 基于实时计算（Flink）打造一个简单的实时推荐系统
    14021
  • 广告场景下的实时计算
    10934
    • 展开全部

    相关文章

  • 关系型数据库mysql的NDB Cluster（或Cluster/NDB）
  • 关系型数据库mysql的CSV
  • 员工上网行为监控中的COBOL遗留系统改造
  • 网络安全与信息安全：漏洞、加密与意识的三重防线
  • 网络安全与信息安全：漏洞、加密与意识的三维防线

    • 相关电子书

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