Flink Streaming：由控制流控制的数据流
我有一个问题是这个问题的变体：Flink：如何存储状态和在另一个流中使用？
我有两条流：
- val ipStream:DataStream[IP地址]=
- <代码>val routeStream：数据流[路由表]=
我想知道哪个包裹使用哪条路线。通常可以通过以下方式完成：
val ip = IPAddress("10.10.10.10")
val table = RoutingTable(Seq("10.10.10.0/24", "5.5.5.0/24"))
val route = table.lookup(ip) // == "10.10.10.0/24"
这里的问题是，我无法在这里真正为流设置密钥，因为这既需要完整的表，也需要ip地址（并且密钥必须独立计算）。
对于ipStream中的每个元素，我需要最新的routeStream元素。现在我正在使用一个所有这些都是非并行处理的黑客：
ipStream
.connect(routeStream)
.keyBy(_ => 0, _ => 0)
.flatMap(new MyRichCoFlatMapFunction) // with ValueState[RoutingTable]
这听起来像是广播策略的用例。但是，routeStream将被更新，并且不会固定在文件中。问题仍然存在：是否有一种方法可以拥有两个流，其中一个流包含另一个流的更改控制数据？
共有1个答案
既然我解决了这个问题，我不妨在这里写一个答案：）
我按如下方式键入了两条流：
- RoutingTable流使用网络路由的第一个字节进行键控
- IP地址也由地址的第一个字节键控
这在IP包通常在网络中使用相同的/8前缀进行路由的情况下工作，这可以假设用于大多数流量。
然后，通过有状态的RichCoFlatMap，可以将路由表状态建立为键。接收新IP包时，请在路由表中进行查找。现在有两种可能的情况：
- 未找到匹配的路由。我们可以将该包存储在此处以备稍后使用，但丢弃也可以
这样，我们有两个流，其中一个流具有另一个流的更改控制数据。
-
截止到现在，在我们所看过的程序中，总是有一系列语句从上到下精确排列，并交由 Python 忠实地执行。如果你想改变这一工作流程，应该怎么做？就像这样的情况：你需要程序作出一些决定，并依据不同的情况去完成不同的事情，例如依据每天时间的不同打印出 ‘早上好’ ‘Good Morning’ 或 ‘晚上好’ ‘Good Evening’？ 正如你可能已经猜测到的那番，这是通过控制流语句来实现的。在 Pyt
-
Swift 提供所有多样化的控制流语句。包括 while 循环来多次执行任务； if ， guard 和 switch 语句来基于特定的条件执行不同的代码分支；还有比如 break 和 continue 语句来传递执行流到你代码的另一个点上。 Swift 同样添加了 for-in 循环，它让你更简便地遍历数组、字典、范围和其他序列。 Swift 的 switch 语句同样比 C 中的对应语句多了不
-
本页包含内容： For 循环 While 循环 条件语句 控制转移语句（Control Transfer Statements） Swift提供了类似 C 语言的流程控制结构，包括可以多次执行任务的for和while循环，基于特定条件选择执行不同代码分支的if和switch语句，还有控制流程跳转到其他代码的break和continue语句。 除了 C 语言里面传统的 for 条件递增（for-co
-
我有一个关于kafka流应用程序中的控制流的基本问题。如果有两个源主题 我做了一个非常初步的测试，当记录被消费时，我偷看了一下，然后用一个简单的速溶软件打印了它们被处理的瞬间。现在 这些是主题中记录的开始和结束时间戳 主题B记录在主题A之前提取。Sysout显示主题B中的所有记录。有人能帮助理解这一点吗？我希望在编写具有多个输入源的流式应用程序时使用这种理解。 提前感谢
-
程序主要有顺序、分支和循环几种执行流程。本节主要讨论如何将Go语言的控制流比较直观地转译为汇编程序，或者说如何以汇编思维来编写Go语言代码。 3.5.1 顺序执行 顺序执行是我们比较熟悉的工作模式，类似俗称流水账编程。所有不含分支、循环和goto语句，并且没有递归调用的Go函数一般都是顺序执行的。 比如有如下顺序执行的代码： func main() { var a = 10 pri
-
任何编程语言都包含的一个必要部分就是改变控制流程：if/else，for等。让我们讲述 Rust 语言中 的这部分内容。