Spark Структурированный потоковый поток

Question

Сценарий немного отличается от классического объединения поток-поток

streamA : поток транзакций: transTS, userid, productid, ...

streamB : поток созданных новых продуктов: productid, productname, createTS, ...)

Я хочу объединить транзакции с productIds, но не могу найти комбинацию водяных знаков / условий объединения, чтобы сделатьэто случилось.

streamA_wm = streamA.withWatermark("transTS", "3 minutes")
streamB_wm = streamB.withWatermark("createTS", "1 day")

streamA_wm
   .join(streamB_wm, "productId AND transTS >= createTS", "leftOuter")

Результат пуст.

Что я делаю не так?

Answer 1

Я думаю, что у вас может неправильный подход. Хотя продукты являются транзакционными при их создании и обновлении, они представляют собой метаданные относительно другого потока транзакций.

Я бы предложил следующее:

1. Присоедините поток транзакций к справочной информации о продуктах, которая не подвергается обработке потоком.
2. Не кэшируйте Продукты, это гарантирует, что вы перейдете к источнику.
3. Используйте паркет, KUDU для продуктов.

Но, возможно, есть причина для потока для продуктов, но ... Что произойдет, если больше не будет выполнено обновление продукта, и вы снова получите данные для этого продукта через поток для транзакций?