KSQL：将多个子记录附加到父记录

我正在尝试使用KSQL（作为confluent-5.0.0的一部分）从一组父记录和子记录中创建单个记录，其中每个父记录都有多个子记录（特别是付款细节和涉及的各方）在付款） . 这些父/子记录由父母的id链接 . 为了说明，我正在处理源系统中大致这种结构的记录：

payment:
| id    | currency | amount | payment_date |
|------------------------------------------|
| pmt01 | USD      | 20000  | 2018-11-20   |
| pmt02 | USD      | 13000  | 2018-11-23   |

payment_parties:
| id    | payment_id | party_type   | party_ident | party_account |
|-----------------------------------------------------------------|
| prt01 | pmt01      | sender       | XXYYZZ23    | (null)        |
| prt02 | pmt01      | intermediary | AADDEE98    | 123456789     |
| prt03 | pmt01      | receiver     | FFGGHH56    | 987654321     |
| prt04 | pmt02      | sender       | XXYYZZ23    | (null)        |
| prt05 | pmt02      | intermediary | (null)      | (null)        |
| prt06 | pmt02      | receiver     | FFGGHH56    | 987654321     |

这些记录以AVRO格式加载到使用Oracle Golden Gate的一组Kafka主题中，每个表都有一个主题 . 这意味着存在以下主题： src_payment 和 src_payment_parties . 按照源系统的运行方式，这些记录的时间戳在几毫秒内 .

现在，目的是将这些记录“压扁”为单个记录，这些记录将从传出的主题中消耗 . 为了说明，对于上面的记录，所需的输出将是这些行：

payment_flattened:
| id    | currency | amount | payment_date | sender_ident | sender_account | intermediary_ident | intermediary_account | receiver_ident | receiver_account |
|----------------------------------------------------------------------------------------------------------------------------------------------------------|
| pmt01 | USD      | 20000  | 2018-11-20   | XXYYZZ23     | (null)         | AADDEE98           | 123456789            | FFGGHH56       | 987654321        |
| pmt02 | USD      | 13000  | 2018-11-23   | XXYYZZ23     | (null)         | (null)             | (null)               | FFGGHH56       | 987654321        |

我想在这里问的第一个问题如下： How can I best achieve this combination of data from the source topics?

当然，我自己也尝试过一些动作 . 为了简洁起见，我将描述我试图将第一个付款方附加到付款记录中 .

第一步：设置源流
注意：由于OGG设置将一个名为'table'的属性添加到AVRO模式，我必须指定要从主题中获取的字段 . 另外，我对指定操作类型的字段（例如插入或更新）不感兴趣 .

create stream payment_stream (id varchar, currency varchar, amount double, \
payment_date varchar) with (kafka_topic='src_payment',value_format='avro');

create stream payment_parties_stream (id varchar, payment_id varchar, party_type varchar, \
party_ident varchar, party_account varchar) with (kafka_topic='src_payment_parties',\
value_format='avro');

第二步：为付款发件人创建流
注意：从我从文档中收集到的，并从实验中发现，为了能够将支付流加入支付方流，后者需要按支付ID进行分区 . 此外，我使联接工作的唯一方法是重命名列 .

create stream payment_sender_stream as select payment_id as id, party_ident, \
party_account from payment_parties_stream where party_type = 'sender' partition by id;

第三步：加入两个流
注意：我正在使用左连接，因为并非每个付款都有所有参与方 . 如上面的示例记录所示， pmt02 没有中介 .

create stream payment_with_sender as select pmt.id as id, pmt.currency, pmt.amount, \
pmt.payment_date, snd.party_ident, snd.party_account from payment_stream pmt left join \
payment_sender_stream snd within 1 seconds on pmt.id = snd.id;

现在，我希望从这个流输出的输出是这样的：

ksql> select * from payment_with_sender;
rowtime | pmt01 | pmt01 | USD | 20000 | 2018-11-20 | XXYYZZ23 | null
rowtime | pmt02 | pmt02 | USD | 13000 | 2018-11-23 | XXYYZZ23 | null

相反，我看到的输出是这样的：

ksql> select * from payment_with_sender;
rowtime | pmt01 | pmt01 | USD | 20000 | 2018-11-20 | null | null
rowtime | pmt01 | pmt01 | USD | 20000 | 2018-11-20 | XXYYZZ23 | null
rowtime | pmt02 | pmt02 | USD | 13000 | 2018-11-23 | null | null
rowtime | pmt02 | pmt02 | USD | 13000 | 2018-11-23 | XXYYZZ23 | null

因此，我想问的第二个（两部分）问题是： Why does the left join produce these duplicate records? And can this be avoided?

对于文本墙的道歉，我试图在问题的描述中尽可能完整 . 当然，我很乐意添加任何可能的遗漏信息，并尽我所知回答有关设置的问题 .