flink问题之Application 模式下启动失败如何解决

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: Apache Flink是由Apache软件基金会开发的开源流处理框架,其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。

问题一:flink job 跑一段时间 watermark 不推进的问题


flink job 跑一段时间 watermark 不推进,任务没挂,source 是 kafka ,kafka 各个partition 均有数据, flink job statue backend 为 memory 。有debug 的姿势推荐吗? 看过 CPU GC 等指标,看不出来有异常。


参考回答:

Flink

metrics里有一项是task相关的指标currentWatermark,从中可以知道subtask_index,task_name,watermark三项信息,应该能帮助排查watermark的推进情况。


关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/370126


问题二:flink1.11 Application 模式下启动失败问题


我把作业提交模式从 yarn-cluster 换成 application 模式,启动失败,报两个错误: 1、java.lang.ClassNotFoundException: org.apache.hadoop.yarn.api.records.ResourceInformation 2、cannot assign instance of org.apache.commons.collections.map.LinkedMap to field org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.pendingOffsetsToCommit of type org.apache.commons.collections.map.LinkedMap in instance of com.tydic.tysc.core.flink.cal.v3.core.connector.kafka.source.KafkaTableSource$CustomerFlinkKafkaConsumer 在 yarn-cluster 下正常运行,请各位帮忙看下。


参考回答:

Application和Perjob模式本质上的区别是用户main运行的位置不一样,所以会导致JM启动的classpath

也不太一样的


关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/370125


问题三:flink sqlUpdate,如何获取里面的字段,字段类型,with 等3个属性


我用flink 注册一张表: CREATE TABLE dim_mysql ( id int, -- type varchar -- ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://localhost:3390/test', 'table-name' = 'flink_test', 'driver' = 'com.mysql.cj.jdbc.Driver', 'username' = '', 'password' = '', 'lookup.cache.max-rows' = '5000', 'lookup.cache.ttl' = '1s', 'lookup.max-retries' = '3' ) 有没有通过 tableEnv 去获取,字段[id,type] 类型[INTEGER,VARCHAR] 以及属性,map 这种。 我看阿里官方有blink 支持自定义sink: publicabstractclassCustomSinkBaseimplementsSerializable{ protectedMap userParamsMap;// 您在sql with语句中定义的键值对,但所有的键均为小写 protectedSet primaryKeys;// 您定义的主键字段名 protectedList headerFields;// 标记为header的字段列表 protectedRowTypeInfo rowTypeInfo;// 字段类型和名称 核心需求是:获取定义的表的所有属性,自己实现自己的功能,包括 join sink 等各种逻辑


参考回答:

tableEnv 中 可以通过

tableEvn.from(xx).getSchema() 拿到该表的schema信息,但是没法拿到对应的properties。

如果要拿到properties,可以通过catalog的接口得到 [1]。

如果要自定义实现source/sink,可以参考 [2]

[1]

https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/catalogs.html

[2]

https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/sourceSinks.html


关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/370124


问题四:flink 1.11 executeSql查询kafka表print没有输出


本人当前使用flink版本1.11.0,但是在执行executeSql后,print时没有在console打印出结果(查看kafka是一直有数据产生的), sql如下:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();

StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

Catalog catalog = new HiveCatalog("x", "default", "D:\conf", "1.1.0");

tEnv.registerCatalog("x", catalog);

TableResult execute = tEnv.executeSql("select * from x.ods.ods_binlog_test_trip_create_t_order_1");

execute.print();

建表语句如下:

CREATE TABLE x.ods.ods_binlog_test_trip_create_t_order_1 (

_type STRING,

order_no STRING,

order_time STRING,

dt as TO_TIMESTAMP(order_time),

proctime as PROCTIME(),

WATERMARK FOR dt AS dt - INTERVAL '5' SECOND

) WITH (

'connector.type' = 'kafka',

'connector.properties.bootstrap.servers' = '***',

'connector.properties.zookeeper.connect' = '****',

'connector.version' = 'universal',

'format.type' = 'json',

'connector.properties.group.id' = 'testGroup',

'connector.startup-mode' = 'group-offsets',

'connector.topic' = 'test'

)


参考回答:

1.11的 TableResult.collect() 和 TableResult.print() 方法在流模式下, 都是exactly once语义,需要配置checkpoint才能得到结果。


关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/370123


问题五:flink1.9写权限认证的es6


请问flink如何将数据写入到权限认证的es集群哪,没找到配置用户名密码的地方,哪位大佬帮忙解答一下


参考回答:

SQL添加认证的逻辑已经在FLINK-18361[1] 中完成了,1.12版本会支持这个功能

[1] https://issues.apache.org/jira/browse/FLINK-18361


关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/372088


相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
2月前
|
分布式计算 资源调度 大数据
大数据-110 Flink 安装部署 下载解压配置 Standalone模式启动 打包依赖(一)
大数据-110 Flink 安装部署 下载解压配置 Standalone模式启动 打包依赖(一)
74 0
|
2月前
|
分布式计算 资源调度 大数据
大数据-110 Flink 安装部署 下载解压配置 Standalone模式启动 打包依赖(二)
大数据-110 Flink 安装部署 下载解压配置 Standalone模式启动 打包依赖(二)
78 0
|
2月前
|
资源调度 分布式计算 大数据
大数据-111 Flink 安装部署 YARN部署模式 FlinkYARN模式申请资源、提交任务
大数据-111 Flink 安装部署 YARN部署模式 FlinkYARN模式申请资源、提交任务
120 0
|
4月前
|
Java 关系型数据库 MySQL
实时计算 Flink版操作报错合集之在使用批处理模式中使用flat_aggregate函数时报错,该如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
4月前
|
资源调度 关系型数据库 MySQL
【Flink on YARN + CDC 3.0】神操作!看完这篇教程,你也能成为数据流处理高手!从零开始,一步步教会你在Flink on YARN模式下如何配置Debezium CDC 3.0,让你的数据库变更数据瞬间飞起来!
【8月更文挑战第15天】随着Apache Flink的普及,企业广泛采用Flink on YARN部署流处理应用,高效利用集群资源。变更数据捕获(CDC)工具在现代数据栈中至关重要,能实时捕捉数据库变化并转发给下游系统处理。本文以Flink on YARN为例,介绍如何在Debezium CDC 3.0中配置MySQL连接器,实现数据流处理。首先确保YARN上已部署Flink集群,接着安装Debezium MySQL连接器并配置Kafka Connect。最后,创建Flink任务消费变更事件并提交任务到Flink集群。通过这些步骤,可以构建出从数据库变更到实时处理的无缝数据管道。
389 2
|
4月前
|
关系型数据库 MySQL 数据处理
实时计算 Flink版产品使用问题之mini-cluster模式下,怎么指定checkpoint的时间间隔
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
4月前
|
资源调度 算法 Java
Flink四种集群模式原理
Flink四种集群模式原理
145 0
|
4月前
|
消息中间件 SQL Kafka
实时计算 Flink版产品使用问题之如何将changelog转换为append模式
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
4月前
|
关系型数据库 MySQL Serverless
实时计算 Flink版产品使用问题之原生Session模式下遇到classpath路径未生效,该怎么办
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
5月前
|
SQL Oracle 关系型数据库
实时计算 Flink版操作报错合集之向远端flink提交cdc模式作业时,连接池中的连接超时,该如何处理
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。

相关产品

  • 实时计算 Flink版