Flink-Netty-Connector??Flink-Netty-ConnectorSocket??
??Flink-Netty-Connector??
您好,这次发邮件主要想请教一下关于Flink Kafka Connector的一些问题:
1、在官方文档中有提到关闭checkpoint,flink仍会定时提交offset。但经过测试和观察源码,发现只有FlinkKafkaConsumer08会在createFetcher时就创建一个定期提交offset的committer,而FlinkKafkaConsumer09和FlinkKafkaConsumer010似乎只能依赖commitInternalOffsetsToKafka()方法提交Offset,该方法只有在checkpoint打开的情况才会被调用。这是否意味着使用FlinkKafk
在Flink 1.8.0下,通过 env.readFile 实时监控目录下的新文件并处理。5千多个文件,有25个文件被遗漏。
逻辑如下:
1、一个Flink程序实时将小文件写入目录A
2、另一个Flink程序通过env.readFile、PROCESS_CONTINUOUSLY模式实时监控目录A,然后做其他操作
发现,第二个Flink程序偶尔会遗漏文件。
请教下: 为什么会有文件丢失,丢失的原因可能是什么?并行度?
Hi 鑫铉:
我尝试解答下;
1. 这是否意味着使用FlinkKafkaConsumer09必须开启checkpoint机制,不然不会定期提交offset至kafka呢?
根据官方文档 [1],checkpoint offset是Flink的功能,auto commit offset是kafka
client的功能;这俩功能在作用上有一定重叠,所以根据配置会有不同的行为表现;
如果Flink没有开启checkpoint,那么offset的保存依赖于kafka client的auto commit offset;
如果Flink开启了checkpoint,那么au
大家好:
我在使用flink table api 实现group by
聚合操作的时候,自定义了一个UDAF函数,首次在集群上运行的时候是正确的,但是从check
point恢复任务的时候报下面错误,但是使用内置的函数就没问题,不知道我该怎么解决呢?
java.lang.RuntimeException: Error while getting state
at
org.apache.flink.runtime.state.DefaultKeyedStateStore.getState(DefaultKeyedStateStore.java:62)
at
org.apach
Hi Victor:
您的回复已收到,谢谢您详细的解答!非常感谢!
Victor Wong 于2019年8月23日周五 上午10:20写道:
> Hi 鑫铉:
> 我尝试解答下;
>
> 1. 这是否意味着使用FlinkKafkaConsumer09必须开启checkpoint机制,不然不会定期提交offset至kafka呢?
> 根据官方文档 [1],checkpoint offset是Flink的功能,auto commit offset是kafka
> client的功能;这俩功能在作用上有一定重叠,所以根据配置会有不同的行为表现;
> 如果Flink没有
flink1.9.0 下 本地web 页面主页404,代码:
```
var config = new Configuration()
config.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true)
config.setInteger(RestOptions.PORT, 8089)
val env = StreamExecutionEnvironment.createLocalEnvironment(8, config)
```
打开 http://localhost:8089/ 显示
{"errors":["Not found
你是在哪看到这个配置的,我查了下代码甚至这个选项都没有使用点(x
Best,
tison.
hb <343122...@163.com> 于2019年8月23日周五 下午1:22写道:
> flink1.9.0 下 本地web 页面主页404,代码:
> ```
> var config = new Configuration()
> config.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true)
>
> config.setInteger(RestOptions.PORT, 8089)
> val env = Strea
1.9 版本之前,都是可以这么用的,正常的,1.9也是有这个API的啊
在 2019-08-23 12:28:14,"Zili Chen" 写道:
>你是在哪看到这个配置的,我查了下代码甚至这个选项都没有使用点(x
>
>Best,
>tison.
>
>
>hb <343122...@163.com> 于2019年8月23日周五 下午1:22写道:
>
>> flink1.9.0 下 本地web 页面主页404,代码:
>> ```
>> var config = new Configuration()
>> config.setBoolean(ConfigConstants.LOCAL
我切到了 1.9 的代码上看,这个 ConfigConstants.LOCAL_START_WEBSERVER 应该是个废设置,设不设都没有任何效果。
所以问题应该跟这个选项没关系,比如你刷新 localhost:8089 能不能恢复?
Best,
tison.
hb <343122...@163.com> 于2019年8月23日周五 下午1:47写道:
> 1.9 版本之前,都是可以这么用的,正常的,1.9也是有这个API的啊
> 在 2019-08-23 12:28:14,"Zili Chen" 写道:
> >你是在哪看到这个配置的,我查了下代码甚至这个选项都没有使用点(x
另外有个问题是,你是下载二进制 zip 还是从源码编译安装的?
Best,
tison.
Zili Chen 于2019年8月23日周五 下午2:04写道:
> 我切到了 1.9 的代码上看,这个 ConfigConstants.LOCAL_START_WEBSERVER 应该是个废设置,设不设都没有任何效果。
>
> 所以问题应该跟这个选项没关系,比如你刷新 localhost:8089 能不能恢复?
>
> Best,
> tison.
>
>
> hb <343122...@163.com> 于2019年8月23日周五 下午1:47写道:
>
>> 1.9 版本之前,都是可以
我在idea里 用maven 下载依赖,在idea里运行flink程序源码里没有标志为废弃啊
```package org.apache.flink.configurationpublic
final class ConfigConstants { ... /** * @deprecated
Use {@link ResourceManagerOptions#LOCAL_NUMBER_RESOURCE_MANAGER} instead.
*/@Deprecatedpublic static final int
DEFA
我在idea里 用maven 下载依赖,在idea里运行flink程序
源码里没有标志为废弃啊
```
package org.apache.flink.configuration
public final class ConfigConstants { ...
/**
* @deprecated Use {@link ResourceManagerOptions#LOCAL_NUMBER_RESOURCE_MANAGER}
instead.
*/
@Deprecated
public static final int DEFAULT_LOCAL_NUMBER_RESOURCE
源码里没有标为废弃应该是 FLINK 的一个 issue,你可以到 JIRA[1]上提,这个选项确实是没有用的。
听起来你的程序是个测试程序,能提供相应的源码吗?如果你说后面能访问 /taskmanagers 的话可能并没有问题(x
Best,
tison.
[1] https://issues.apache.org/jira/browse/
hb <343122...@163.com> 于2019年8月23日周五 下午2:27写道:
> 我在idea里 用maven 下载依赖,在idea里运行flink程序源码里没有标志为废弃啊
> ```package
> org.apach
个人理解,能通过这个端口访问restfull API 应该表示 这个设置生效了,但是首页却是404
访问: http://localhost:8089/ 404
访问: http://localhost:8089/taskmanagers/
正常:
{"taskmanagers":[{"id":"ef7030d2-eb13-4c68-8d0c-20b3d59616c8","path":"akka://flink/user/taskmanager_0","dataPort":-1,"timeSinceLastHeartbeat":0,"slotsNumber":8,"freeSl
15 matches
Mail list logo