你好，我的Flink作业反压严重，请问如何判断瓶颈点在哪？如何确定是资源不足还是其他问题？

我的Flink作业提交报错了，我咋知道哪里有问题，提工单处理太慢了。

【功能模块】flink执行yarn-session报错【操作步骤&问题现象】1、MRS版本为8.1.0.1flink版本为1.12.0根据上个版本的flink组件配置进行配置之后，执行命令yarn-session报错，具体报错信息如下：2、完整信息如文档所示【截图信息】【日志信息】（可选，上传日志内容或者附件）

请问下DLI Flink作业支持哪些数据格式和数据源?我想使用json输入，parquet输出，请问如何实现

我使用Flink Opensource SQL, 采用Flink 1.10版本。初期Flink作业规划的Kafka的分区数partition设置过小或过大，后期需要更改Kafka区分数，很麻烦。如何不停止作业实现自动感知？

【功能模块】DLI Flink【操作步骤&问题现象】请教下Flink作业重启后，如何保证不丢失数据？

【功能模块】DLI Flink【操作步骤&问题现象】我现在Kafka的数据流量是两千万records/s，我需要配置多少个taskmanager，jobmanager的spu设置为多少？

【功能模块】DLI Flink作业如何配置资源？【操作步骤&问题现象】这些参数到底怎么配置，怎么计算？【截图信息】【日志信息】（可选，上传日志内容或者附件）

【功能模块】DLI Flink【操作步骤&问题现象】使用DLI Flink发现页面上有反压，数据消费变慢了【截图信息】【日志信息】（可选，上传日志内容或者附件）

【功能模块】Flink【操作步骤&问题现象】你好，我看资料里有opensource语法的资料，但直接使用会报语法错误，请问如何才能使用。【截图信息】【日志信息】（可选，上传日志内容或者附件）

【功能模块】DLI Flink【操作步骤&问题现象】1、使用DLI Flink Jar作业看不到日志输出2、从转储的日志来看，日志全输出到.err日志文件中，没有输出到.log文件中【截图信息】【日志信息】（可选，上传日志内容或者附件）

【求助】 1.任务提交到集群是使用的配置是集群上的flink/conf/下的日志配置文件 2.如何在任务提交时指定自定义的配置文件？

【功能模块】Flink DataStream 数据sink到带有用户名密码的验证的ES集群中报错。【操作步骤&问题现象】1、使用开源的flink-connector-elasticsearch6_2.11 的jar包2、SunCertPathBuilderException: unable to find valid certification path to requested target3、在es客户端那 提示找不到有效的证书路径  我觉得应该是协议错误 我们集群使用curl 方式操作es 需要指定  --tlsv1.2【求助】1.是不是不能使用开源的flink 连接器 sink 到es2.有没有FI集群sink 到ES的样例demo 可以参考一下3.或者帮忙提醒一下从哪里设置tlsv1.2协议  【日志信息】（可选，上传日志内容或者附件）Caused by: org.elasticsearch.client.ResponseException: method [HEAD], host [https://10.28.132.195:24100], URI [/], status line [HTTP/1.1 403 Forbidden]

### 1. 首先客户需要在消息通知服务（SMN）中提前创建一个【主题】，并将客户指定的邮箱或者手机号添加到主题订阅中。这时候指定的邮箱或者手机会收到请求订阅的通知，点击链接确认订阅即可。   ### 2. 创建Flink SQL作业，编写作业SQL完成后，配置【运行参数】。 #### 2.1 配置作业的【CU数量】、【管理单元】与【最大并行数】，依据如下公式： ```sql CU数量 = 管理单元 + (算子总并行数 / 单TM Slot数) * 单TM所占CU数 ``` 例如：CU数量为9CU，管理单元为1CU，最大并行数为16，则计算单元为8CU。 如果不手动配置TaskManager资源，则单TM所占CU数默认为1，单TM slot数显示值为0，实际值依据上述公式计算结果为 16÷(9-1)=2。 #### 2.2 勾选【保存作业日志】按钮，选择一个OBS桶。如该桶未授权，需点击【立即授权】。此项配置可以在作业异常失败后将作业日志保存到客户的OBS桶下，方便客户定位故障原因。  #### 2.3 勾选【作业异常告警】选项，选择前述步骤创建的【SMN主题】。此项配置可以在作业异常情况下，向客户指定邮箱或者手机发送消息通知，方便客户及时感知异常。  #### 2.4 勾选【开启Checkpoint】选项，依据自身业务情况调整Checkpoint间隔和模式。Flink checkpoint机制可以保证Flink任务突然失败时，能够从最近的Checkpoint进行状态恢复重启。  说明： - 此处的Checkpoint间隔为两次触发Checkpoint的间隔，执行checkpoint会影响实时计算性能，配置间隔时间需权衡对业务的性能影响及恢复时长，最好大于Checkpoint的完成时间，建议设置为5min。 - Exactly Once模式保证每条数据只被消费一次，At Least Once模式每条数据至少被消费一次，请依据业务情况选择。 #### 2.5 勾选【异常自动恢复】与【从Checkpoint恢复】，根据自身业务情况选择重试次数。 #### 2.6 配置【脏数据策略】，依据自身的业务逻辑和数据特征选择忽略、抛出异常或者保存脏数据。 选择【运行队列】，提交并运行作业。 Flink Jar作业可靠性配置与SQL作业相同，不再另行说明。 ### 3. 登录【云监控服务CES】，在【云服务监控】列表中找到【数据湖探索】服务，在Flink作业中找到目标作业，点击【创建告警规则】。    DLI 为Flink作业提供了丰富的监控指标，客户可以依据自身需求使用不同的监控指标定义告警规则，实现更细粒度的作业监控。 监控指标说明见：https://support.huaweicloud.com/usermanual-dli/dli_01_0445.html

【功能模块】版本信息BASE    6.5.1    ( 补丁  6.5.1.7 )Flink    6.5.1    ( 补丁  6.5.1.7 )Porter    6.5.1    ( 补丁  6.5.1.7 )Spark2x    6.5.1    ( 补丁  6.5.1.7 )HD    6.5.1    ( 补丁  6.5.1.7 )Elasticsearch    6.5.1    ( 补丁  6.5.1.7 )在flink客户端下执行  命令如下bin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka /opt/client/flinkStreamKafka.jar --topic Auto_Agreement --bootstrap.servers ip:21007 --security.protocol SASL_PLAINTEXT --sasl.kerberos.service.name kafka --kerberos.domain.name hadoop.hadoop.combin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka /opt/tmpdata/sou1yu/flinkStreamKafka.jar --topic Auto_Agreement --bootstrap.servers ip:21005【操作步骤&问题现象】1、正常导入项目 导入jar包2、source 过执行环境  kinit 过用户 3.在flink客户端下执行  命令如下 都不行（下面命令的ip也正确的 下面只是为了保护一下ip所以处理了一下）bin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka /opt/client/flinkStreamKafka.jar --topic Auto_Agreement --bootstrap.servers ip:21007 --security.protocol SASL_PLAINTEXT --sasl.kerberos.service.name kafka --kerberos.domain.name hadoop.hadoop.combin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka /opt/tmpdata/sou1yu/flinkStreamKafka.jar --topic Auto_Agreement --bootstrap.servers ip:21005【截图信息】【日志信息】（可选，上传日志内容或者附件）SLF4J: Class path contains multiple SLF4J bindings.SLF4J: Found binding in [jar:file:/opt/client/Flink/flink/lib/logback-classic-1.2.3.jar!/org/slf4j/impl/StaticLoggerBinder.class]SLF4J: Found binding in [jar:file:/opt/client/Flink/flink/lib/flink-dist_2.11-1.7.2.jar!/org/slf4j/impl/StaticLoggerBinder.class]SLF4J: Found binding in [jar:file:/opt/client/HDFS/hadoop/share/hadoop/common/lib/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.SLF4J: Actual binding is of type [ch.qos.logback.classic.util.ContextSelectorStaticBinder]Starting execution of programuse command as: ./bin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka /opt/test.jar --topic topic-test -bootstrap.servers xxx.xxx.xxx.xxx:21005./bin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka /opt/test.jar --topic topic-test -bootstrap.servers xxx.xxx.xxx.xxx:21007 --security.protocol SASL_PLAINTEXT --sasl.kerberos.service.name kafka./bin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka /opt/test.jar --topic topic-test -bootstrap.servers xxx.xxx.xxx.xxx:21008 --security.protocol SSL --ssl.truststore.location /home/truststore.jks --ssl.truststore.password huawei./bin/flink run --class com.huawei.bigdata.flink.examples.WriteIntoKafka /opt/test.jar --topic topic-test -bootstrap.servers xxx.xxx.xxx.xxx:21009 --security.protocol SASL_SSL --sasl.kerberos.service.name kafka --ssl.truststore.location /home/truststore.jks --ssl.truststore.password huawei******************************************************************************************<topic> is the kafka topic name<bootstrap.servers> is the ip:port list of brokers******************************************************************************************------------------------------------------------------------ The program finished with the following exception:org.apache.flink.client.program.ProgramInvocationException: Could not retrieve the execution result. (JobID: 1a0dc0071081418158fbb38a8326a920)        at org.apache.flink.client.program.rest.RestClusterClient.submitJob(RestClusterClient.java:261)        at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:490)        at org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:66)        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1510)        at com.huawei.bigdata.flink.examples.WriteIntoKafka.main(WriteIntoKafka.java:32)        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)        at java.lang.reflect.Method.invoke(Method.java:498)        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529)        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421)        at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:430)        at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:814)        at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:288)        at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:213)        at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1051)        at org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1127)        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.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)        at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1127)Caused by: org.apache.flink.runtime.client.JobSubmissionException: Failed to submit JobGraph.        at org.apache.flink.client.program.rest.RestClusterClient.lambda$submitJob$8(RestClusterClient.java:391)        at java.util.concurrent.CompletableFuture.uniExceptionally(CompletableFuture.java:870)        at java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(CompletableFuture.java:852)        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)        at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)        at org.apache.flink.runtime.concurrent.FutureUtils.lambda$retryOperationWithDelay$5(FutureUtils.java:203)        at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)        at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)        at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)        at org.apache.flink.runtime.concurrent.FutureUtils$Timeout.run(FutureUtils.java:795)        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)        at java.util.concurrent.FutureTask.run(FutureTask.java:266)        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)        at java.lang.Thread.run(Thread.java:748)Caused by: java.util.concurrent.TimeoutException        ... 8 more