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 86 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吗?
54
1
0
为什么使用FlinkCDC和Kakfa Sink向Kakfa中发送binlog数据，会存在脏数据？
42
0
0
canal -adapter只能适配mysql库binlog_format=row模式？
40
1
0
FLinkcdc 在初始化快照阶段的时候是不是就会记录下binlog点位？
39
1
0
Flink的holo表的binlog最长生命周期是多长啊？
38
1
0
Flink的datastream api消费holo表的binlog的时候起始时间是怎么设置的？
37
1
0
Flink多个表的同步顺序 是按照binlog顺序执行的么？有个表的同步失败了会影响后续执行吗？
28
1
0
云数据仓库ADB打开表的binlog特性，对表会有什么影响吗？
47
1
0
flinkcdc 采集mysql出现卡在某个binlog文件出不来怎么办？
43
0
0
flink-cdc 在抽取mysql binlog，运行一段时间 source 就报java.uti
1030
3
0
大数据与机器学习

实时计算 Flink

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

我要提问

热门讨论

热门文章

  • FFA 2024 大会门票免费送！AI时代下大数据技术未来路在何方？
    224
  • Flink CDC这个问题, 有没有大佬遇到过, mysql binlog日志还在, 读取失败？
    551
  • 大佬，Caused by: java.lang.ClassNotFoundException: or
    383
  • flink怎么能够快速消费kafka数据，需要设置什么参数呢？
    1401
  • Flink如何配置Task Manager？
    556
  • 请问flink-connector-jdbc在sink时，如何设置批量写入？
    3047
  • 请问如何用flink sql客户端用yarn application模式提交任务呢？
    1460
  • 请问现在Flink的批作业能定时调度吗？或者通过什么方法才能实现定时调度啊？
    912
  • 使用flink on yarn的模式，怎么进行内存资源调优呢，如何配置flink内存
    653
  • Slot request bulk is not fulfillable，20以上job同时部署失败
    630
    • 展开全部
  • OPPO数据中台之基石：基于Flink SQL构建实数据仓库
    21902
  • 独家专访阿里集团副总裁贾扬清：我为什么选择加入阿里巴巴？
    17597
  • Flume+Kafka+Flink+Redis构建大数据实时处理系统：实时统计网站PV、UV展示
    23321
  • 阿里云实时计算产品案例&解决方案汇总
    27788
  • 使用 Kafka 和 Flink 构建实时数据处理系统
    12969
  • 【对话科技】Flink技术介绍和新功能展望
    6843
  • Flink: 实时规则引擎助力新零售发展
    9985
  • Flink: 快速构建统一的实时日志平台
    8409
  • Apache Flink 漫谈系列(15) - DataStream Connectors之Kafka
    8895
  • 基于 Flink 构建 CEP 引擎的挑战和实践
    6650
    • 展开全部

    相关文章

  • Mysql高可用架构方案
  • 面试：什么是死锁，如何避免或解决死锁；MySQL中的死锁现象，MySQL死锁如何解决
  • mysql基础知识
  • 如何选择合适的CPU进行维修？
  • 网络安全的屏障与钥匙：漏洞、加密与意识的交织

    • 相关电子书

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