流 DSL

内部联接（窗口化）

• （KStream，KStream） → KStream

将此流与另一个流执行 INNER JOIN。 即使此操作是窗口化的，联接的流也将是 type 而不是 .（详情）KStream<K, ...>KStream<Windowed<K>, ...>

数据必须协同分区：两侧的输入数据必须协同分区。

当且仅当流被标记为重新分区时，才导致流的数据重新分区（如果两个流都被标记，则两个流都被重新分区）。

exists 的几个变体，有关详细信息，请参阅 Javadocs。join

import java.time.Duration;
KStream<String, Long> left = ...;
KStream<String, Double> right = ...;

// Java 8+ example, using lambda expressions
KStream<String, String> joined = left.join(right,
    (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
    Joined.with(
      Serdes.String(), /* key */
      Serdes.Long(),   /* left value */
      Serdes.Double())  /* right value */
  );

详细行为：

• 连接是基于键的，即使用 join predicate ，并且是基于窗口的，即当且仅当它们的 时间戳由 User-Supplied 定义，彼此“接近”，即窗口在记录时间戳上定义一个额外的连接谓词。leftRecord.key == rightRecord.keyJoinWindows

• 每当收到新输入时，都会在下面列出的条件下触发联接。当它被触发时，将调用用户提供的 连接输出记录。ValueJoiner

  • 具有键或值的输入记录将被忽略，并且不会触发联接。nullnull

有关详细说明，请参阅本节底部的语义概述。

外部联接（窗口式）

• （KStream，KStream） → KStream

将此流与另一个流执行 OUTER JOIN。 即使此操作是窗口化的，联接的流也将是 type 而不是 .（详情）KStream<K, ...>KStream<Windowed<K>, ...>

数据必须协同分区：两侧的输入数据必须协同分区。

当且仅当流被标记为重新分区时，才导致流的数据重新分区（如果两个流都被标记，则两个流都被重新分区）。

exists 的几个变体，有关详细信息，请参阅 Javadocs。outerJoin

import java.time.Duration;
KStream<String, Long> left = ...;
KStream<String, Double> right = ...;

// Java 8+ example, using lambda expressions
KStream<String, String> joined = left.outerJoin(right,
    (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
    Joined.with(
      Serdes.String(), /* key */
      Serdes.Long(),   /* left value */
      Serdes.Double())  /* right value */
  );

详细行为：

• 连接是基于键的，即使用 join predicate ，并且是基于窗口的，即当且仅当它们的 时间戳由 User-Supplied 定义，彼此“接近”，即窗口在记录时间戳上定义一个额外的连接谓词。leftRecord.key == rightRecord.keyJoinWindows

• 每当收到新输入时，都会在下面列出的条件下触发联接。当它被触发时，将调用用户提供的 连接输出记录。ValueJoiner

  • 具有值的输入记录将被忽略，并且不会触发联接。null

• 对于一侧没有任何匹配项的一侧的每个 input 记录，将分别使用 或 调用 ;下表中 timestamp=60、timestamp=80 和 timestamp=100 的行对此进行了说明，其中列出了 、 和 OUTER JOIN 列。 请注意，这些 left 和 right 结果将在指定的宽限期过后发出。注意：使用已弃用的 API 可能会导致急切地发出虚假的左/右结果。ValueJoinerValueJoiner#apply(leftRecord.value, null)ValueJoiner#apply(null, rightRecord.value)[E, null][F, null][null, f]JoinWindows.of(...).grace(...)

有关详细说明，请参阅本节底部的语义概述。

内部联接

• （KTable、KTable） → KTable

将此表与另一个表执行 INNER JOIN。 结果是一个不断更新的 KTable，它表示联接的 “当前” 结果。（详情）

数据必须协同分区：两侧的输入数据必须协同分区。

KTable<String, Long> left = ...;
KTable<String, Double> right = ...;

// Java 8+ example, using lambda expressions
KTable<String, String> joined = left.join(right,
    (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue /* ValueJoiner */
  );

详细行为：

• 连接是基于键的，即使用 join 谓词 。leftRecord.key == rightRecord.key

• 每当收到新输入时，都会在下面列出的条件下触发联接。当它被触发时，将调用用户提供的 连接输出记录。ValueJoiner

  • 带有键的输入记录将被忽略，并且不会触发联接。null
  • 具有值的输入记录将解释为相应键的逻辑删除，这表示从表中删除了键。墓碑不会 触发联接。当收到输入逻辑删除时，如果需要，输出逻辑删除将直接转发到连接结果 KTable（即，仅当相应的 key 实际上已经存在于连接结果 KTable 中）。null
  • 连接版本化表时， 无序输入记录，即同一表中的另一条记录具有相同键和更大时间戳的记录， 已被处理，将被忽略并且不会触发联接。

有关详细说明，请参阅本节底部的语义概述。

外部联接

• （KTable、KTable） → KTable

将此表与另一个表执行 OUTER JOIN。（详情）

数据必须协同分区：两侧的输入数据必须协同分区。

KTable<String, Long> left = ...;
KTable<String, Double> right = ...;

// Java 8+ example, using lambda expressions
KTable<String, String> joined = left.outerJoin(right,
    (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue /* ValueJoiner */
  );

详细行为：

• 连接是基于键的，即使用 join 谓词 。leftRecord.key == rightRecord.key

• 每当收到新输入时，都会在下面列出的条件下触发联接。当它被触发时，将调用用户提供的 连接输出记录。ValueJoiner

  • 带有键的输入记录将被忽略，并且不会触发联接。null
  • 具有值的输入记录将解释为相应键的逻辑删除，这表示从表中删除了键。逻辑删除可能会触发连接， 取决于 Left 和 Right 表中的内容。当收到输入逻辑删除时，如果需要，输出逻辑删除将直接转发到连接结果 KTable（即，仅当相应的 key 实际上已经存在于连接结果 KTable 中）。null
  • 连接版本化表时， 无序输入记录，即同一表中的另一条记录具有相同键和更大时间戳的记录， 已被处理，将被忽略并且不会触发联接。

• 对于一侧没有任何匹配项的一侧的每个 input 记录，将分别使用 或 调用 ;这解释了下表中 timestamp=3 和 timestamp=7 的行，它们分别在 OUTER JOIN 列中列出 和 。ValueJoinerValueJoiner#apply(leftRecord.value, null)ValueJoiner#apply(null, rightRecord.value)[A, null][null, b]

有关详细说明，请参阅本节底部的语义概述。

内部联接

• （KTable、KTable） → KTable

执行此 table 替换为另一个 table。结果是不断更新 KTable 的 URL 来表示联接的 “当前” 结果。（详情）

KTable<String, Long> left = ...;
                KTable<Long, Double> right = ...;
//This foreignKeyExtractor simply uses the left-value to map to the right-key.
Function<Long, Long> foreignKeyExtractor = (x) -> x;

// Java 8+ example, using lambda expressions
                KTable<String, String> joined = left.join(right, foreignKeyExtractor,
                    (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue /* ValueJoiner */
                  );

详细行为：

• join 是基于键的，即 替换为 Join 谓词：

  foreignKeyExtractor.apply(leftRecord.value) == rightRecord.key

• join 将在 条件，每当新输入为 收到。触发时，将调用用户提供的 join 输出记录。ValueJoiner

  • 其 produces 的记录将被忽略，并且不会触发联接。 如果要使用外键进行联接，请使用合适的 sentinel 值（即 对于 String 字段或自动递增的 Integer 字段）。foreignKeyExtractornullnull"NULL"-1
  • 具有值的输入记录将解释为相应键的逻辑删除，这表示 从表中删除键。墓碑不会 触发联接。当输入 tombstone 接收到，然后输出 Tombstone 直接转发到 Join result KTable（即仅当 相应的 key 实际上已经存在于 连接结果 KTable）。null
  • 连接版本化表时， 无序输入记录，即同一表中的另一条记录具有相同键和更大时间戳的记录， 已被处理，将被忽略并且不会触发联接。

请参阅底部的语义概述 有关详细说明。

内部联接

• （KStream、KTable） → KStream

使用表执行此流的 INNER JOIN，从而有效地执行表查找。（详情）

数据必须协同分区：两侧的输入数据必须协同分区。

当且仅当流被标记为重新分区时，才导致流的数据重新分区。

exists 的几个变体，有关详细信息，请参阅 Javadocs。join

KStream<String, Long> left = ...;
KTable<String, Double> right = ...;

// Java 8+ example, using lambda expressions
KStream<String, String> joined = left.join(right,
    (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
    Joined.keySerde(Serdes.String()) /* key */
      .withValueSerde(Serdes.Long()) /* left value */
      .withGracePeriod(Duration.ZERO) /* grace period */
  );

详细行为：

• 连接是基于键的，即使用 join 谓词 。leftRecord.key == rightRecord.key

• 每当收到新输入时，都会在下面列出的条件下触发联接。当它被触发时，将调用用户提供的 连接输出记录。ValueJoiner

  • 只有左侧 （stream） 的输入记录才会触发联接。右侧（表）的输入记录仅更新内部右侧联接状态。
  • 具有键或值的流的输入记录将被忽略，并且不会触发联接。nullnull
  • 具有值的表的输入记录将解释为相应键的逻辑删除，这表示从表中删除了键。 逻辑删除不会触发联接。null

• 对表进行版本控制后， 要联接的表记录是通过执行带时间戳的查找来确定的，即联接的表记录将是具有时间戳的最新时间戳记录 小于或等于流记录时间戳。如果流记录时间戳早于表的历史记录保留期，则会删除该记录。
• 要使用宽限期，需要对表进行版本控制。 这将导致流在尝试在表中查找具有正确时间戳的匹配记录之前缓冲指定的宽限期。 使用宽限期的情况是，表中记录的时间戳小于或等于流记录时间戳，但在流记录之后到达。 如果表记录在宽限期内到达，则仍将进行联接。 如果表记录在宽限期之前未到达，则联接将照常继续。

有关详细说明，请参阅本节底部的语义概述。

内部联接

• （KStream，GlobalKTable） → KStream

使用全局表执行此流的 INNER JOIN，从而有效地执行表查找。（详情）

在实例（重新）启动时完全引导，这意味着该表完全填充了底层主题中的所有数据，即 在启动时可用。实际的数据处理仅在引导完成后才开始。GlobalKTableKafkaStreams

当且仅当流被标记为重新分区时，才导致流的数据重新分区。

KStream<String, Long> left = ...;
GlobalKTable<Integer, Double> right = ...;

// Java 8+ example, using lambda expressions
KStream<String, String> joined = left.join(right,
    (leftKey, leftValue) -> leftKey.length(), /* derive a (potentially) new key by which to lookup against the table */
    (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue /* ValueJoiner */
  );

详细行为：

• 连接是间接基于键的，即使用 join 谓词 。KeyValueMapper#apply(leftRecord.key, leftRecord.value) == rightRecord.key

• 每当收到新输入时，都会在下面列出的条件下触发联接。当它被触发时，将调用用户提供的 连接输出记录。ValueJoiner

  • 只有左侧 （stream） 的输入记录才会触发联接。右侧（表）的输入记录仅更新内部右侧联接状态。
  • 具有键或值的流的输入记录将被忽略，并且不会触发联接。nullnull
  • 具有值的表的输入记录将解释为逻辑删除，这表示从表中删除了记录键。逻辑删除不会触发 加入。null