我左联有问题，好像不行。内部连接是可以的。我用的是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上处理具有相同密钥的所有消息？我需要组装完整的对象并保存到数据库