大家好：
请问，对一个DataStream重复声明watermark，前面的watermark会被覆盖掉吗？
比如我再source端声明了watermark，进行了一系列操作后，我觉得watermark的延迟不满足需求，就再次声明一次。
另外，稍微咨询下另外一个问题，两个流join之后，watermark会消失吗？看书上说的是，以两个流最小的watermark（全局最小）为准。
主要是在阿里云Blink上，使用sql进行join后，说的是时间属性字段会消失。有点不明白。
Hi 苟刚：
另外，看了你的计算逻辑，是同一份数据源里面，有不同类型的数据需要分开处理，你可以尝试使用Flink的侧输出来做分流处理，这样逻辑更清晰，而且程序效率也会高许多
| |
ss
|
|
yangx...@163.com
|
签名由网易邮箱大师定制
在2020年4月7日 19:54，LakeShen 写道：
Hi 苟刚，
Flink 任务中，如果开启 Checkpoint 的话，会在每次Checkpoint
完成后，提交偏移量。如果没有开启的话，就是根据自动提交来提交偏移量，默认是开启的，间隔是 5 s.
至于你说每次都是重头开始的，我个人的想法是不是在代码中设置了从
Hi 苟刚，
Flink 任务中，如果开启 Checkpoint 的话，会在每次Checkpoint
完成后，提交偏移量。如果没有开启的话，就是根据自动提交来提交偏移量，默认是开启的，间隔是 5 s.
至于你说每次都是重头开始的，我个人的想法是不是在代码中设置了从最早开始消费，也就是 你使用到了这个方法：setStartFromEarliest[1]
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/connectors/kafka.html#kafka-consumers-start-positio
你这是个新思路，分成两个 job，但是感觉不太值当，或许这里是 Flink目前
API或者说编程模型很受限的地方，我只是源头数据来自两个地方，要合并下两个数据源，所有下游处理都是一样的。如果按照 actor 的松散模式，我是可以在两个
SourceActor 之间协调的，一个 SourceActor 发完后，通知另一个 SourceActor 再发，或者启动一个新的
SourceActor，大家都往同一个下游 actor 发消息。
user@flink 里那个哥们提到 InputSelectable，我还没看明白怎么能用到 DataStream上，似乎它只实现在
StreamOpe
如果你的作业没有 state，那么 全量和增量部分，可以分成两个独立的作业。
如果你的作业有 state，主要就是设计 state 复用的问题。两个作业的拓扑结构需要确保一致（为了复用 savepoint），一个作业的
source operator 是 jdbc，另一个 source operator 是 kafka。
当运行完 jdbc 作业后，对作业触发一次 savepoint，然后用这个 savepoint 恢复 kafka 作业，可以从 earliest
开始读取（假设作业支持幂等）。
这里的一个问题，主要是如何对 jdbc 作业触发 savepoint，因为 jdbc Inpu
好的，我试一下，有结果了同步大家，谢谢！
在 2020/4/7
下午3:52，“Evan” 写入:
之前的代码好像乱码了，我设置了一下，重新发一下，建议你
在获取consumer之后，再设置一下 consumer.setStartFromLatest();，这样设置的参考就是官网文档介绍的，这是我之前翻译的，可以看一下后边关于【Kafka
Consumers 从指定位置开始消费】的解释，链接：https://www.jianshu.com/p/b753527b91a6
/**
* @param env
我看了下 streamsql 那个 bootstrap 一段，没看懂怎么处理，我现在已经有 mysql table，算是 materialized view
了，也有一份
Kafka 里的 binlog 数据，问题的麻烦在于怎么「先消费 jdbc table, 消费完后，再切换到 kafka 上」，从 flink 文档来看，一旦
我把 jdbc table 的流和 kafka 的流加入 env 里，env.execute() 之后，两个流就同时往下游发数据了——我期望的是 jdbc
table
的流发完了，才开始发 kafka 的流。
谢谢！
On 2020/4/7, 2:16 P
没有 join，只是简单的 union：
DataStream binlogStream = env.addSource(new FlinkKafkaConsumer(…));
DataStream snapshotStream =
env.createInput(JDBCInputFormat.buildJDBCInputFormat()….);
// map() is to convert two streams into same type: (action, fields…),
where action is “ins
之前的代码好像乱码了，我设置了一下，重新发一下，建议你
在获取consumer之后，再设置一下 consumer.setStartFromLatest();，这样设置的参考就是官网文档介绍的，这是我之前翻译的，可以看一下后边关于【Kafka
Consumers 从指定位置开始消费】的解释，链接：https://www.jianshu.com/p/b753527b91a6
/**
* @param env
* @param topic
* @param time 订阅的时间
* @return
* @throws IllegalA
苟刚你好,刚才看了你的kafka消费代码，建议你在获取consumer后，增加一行如下代码
“consumer.setStartFromLatest();”然后再测试一下。
/**
* @param env
* @param topic
* @param time 订阅的时间
* @return
* @throws IllegalAccessException
*/
public static DataStreamSource
10 matches
Mail list logo
