我左联有问题,好像不行。内部连接是可以的。我用的是streamnative pulsarSpark连接器。
Dataset datasetPerson = ...
Dataset datasetPersonWithWt = datasetPerson.withWatermark("__eventTime", "20 seconds").as("per");
Dataset datasetPhone = ...
Dataset datasetPhoneWithWt = datasetPolicy.withWatermark("__eventTime", "20 seconds").as("ph");
Dataset joined = datasetPersonWithWt.join(
datasetPhoneWithWt,
// functions.expr("per.__key=ph.__key"),
functions.expr("per.__key=ph.__key and ph.__eventTime >= per.__eventTime - interval 20 seconds and ph.__eventTime <= per.__eventTime + interval 20 seconds"),
"left");
查询有什么问题?
然而令人疑惑的是,连接器以独占模式连接到脉冲星。spark是否确保在同一个worker上处理具有相同密钥的所有消息?我需要组装完整的对象并保存到数据库
暂无答案!
目前还没有任何答案,快来回答吧!