《流程》专题

我习惯了用Scala编程，但我必须编写一些Java，我正在尝试执行相当于以下Scala片段的操作： 也就是说，我在

假设我有一个函数，它接受两个参数并返回一个值，那么可以将映射转换为流中的列表作为非终端操作吗？我能找到的最近的方法是使用地图上的forEach创建实例并将其添加到预定义列表中，然后从该列表中启动一个新流。还是我错过了什么？ 经典的“在一长串单词中找出3个最常出现的单词” （现在我想流式处理该地图的入口集）

我有一个akka流，我有一个ADT的形式。 现在我有一个这个消息处理程序流和一个那个消息处理程序流。我有一个接受消息类型的入口流。 为了创建拆分，我有以下分区器。我有以下分区器函数的定义。 我希望使用上述方法，并使用类型params中的ADT来初始化分区程序。 编译器抛出这个错误。 据我所知，分区对象只有Inlet（在本例中为A，参数化类型。 有人知道我该怎么解决这个问题吗？

我有以下简单的case类层次结构： 我有一个（来自一个基于Websocket的协议，已经有了编解码器）。 我想将此解复用为Foo和Baz类型的单独流，因为它们由完全不同的路径处理。 最简单的方法是什么？应该很明显，但我错过了一些东西。。。

我正在使用Kafka Streams，我注意到它使我的kafka日志记录了很多日志消息，例如： 这真的很令人不安，因为我发现它会淹没日志，所以我看不到任何其他内容（也会消耗资源）。 为什么它发生在（一些）Kafka Streams内部主题上，而不是其他主题上？ 我怎样才能禁用它？

我想在不断从客户端接收实时视频流的情况下使用MPEG-DASH技术。Web服务器获取实时视频流，不断生成m4s文件，并在mpd中声明它。因此可以不断播放新片段。（我使用的是FFMPEG的ffserver。因此视频流继续累积在 /tmp/feed1.ffm文件中。） 使用MP4Box似乎能够生成mpd，init。mp4，M4用于现有文件。但它似乎不支持直播。 我想要片段格式的MP4而不是mpeg-t

我正在构建一个事件驱动的微服务架构，它应该是云不可知的（尽可能多）<由于这最初是在GCP中进行的，我不想在配置和所有这些方面花费太长时间，我打算直接将GCP的发布/订阅用于事件队列，并在稍后处理其他云实现，但后来我遇到了Spring云数据流，这看起来很好，因为这些是Spring Boot微服务，我需要一种方法来协调它们 Spring Cloud数据流是否支持Pub Sub作为事件队列？ 在配置和设

我已经建立一个网站一段时间了，我仍然坚持那件事： 我在dbm数据库中为我的网站存储了一些小视频（最多大约400MB），我想在我的网站上播放它们。 我正在使用Tornado python框架手工构建请求处理程序，我想知道如何构建我的处理程序。我从未发现媒体流是如何工作的，也没有在网上找到很多话题。 所以我想要实现的完整结果是在我的网站上有一个网络播放器，在那里我可以请求特定的视频，然后播放它们，而不

我似乎找不到一个像样的例子来说明如何通过Python使用AWS动态流。有人能给我举几个例子吗？ 最好的

有人尝试在Apache Flink中使用DynamoDB流吗？ Flink有一个Kinesis消费者。但是我正在寻找如何直接使用Dynamo流。 我试了很多次，但什么也没找到。然而，Flink Jira董事会发现一个未决请求。所以我想这个选项还不可用？我有什么选择？ 允许FlinkKinesisConsumer适应AWS DynamoDB流

我正在尝试创建一个查找器，它需要几个谓词并减少它们：

如何为每个作业设置流程参数？我试图配置一个自定义警报程序，我希望每个作业都触发它。它看起来像是在从流参数中寻找“alert.type”属性，但现在我只能通过接口触发它。有什么想法吗？

我已经讨论了一些相关问题，比如如何确保java8流中的处理顺序？，我仍然不完全清楚输出元素的顺序。因此，请澄清我的以下疑问。 我认为至少在理论上（或根据java规范），它可以按1、2、3、4、5、6、7、8以外的随机顺序打印。我说得对吗？ 还有一个相关的问题——遭遇订单保存的决定是在执行的哪个点做出的？更准确地说，在执行开始之前，是否通过检查源、中间操作和终端操作的特性来评估整个流水线的订单特性？

我有一个Apache Beam管道，它在读取BigQuery后试图写入Postgres。代码使用JdbcIO连接器和数据流运行器。我使用的是Python 3.8.7和Apache Beam 2.28.0 我使用的是默认扩展服务。我也尝试运行一个自定义扩展服务，但仍然得到相同的错误。你知道吗？ 我得到以下错误

启动日志的尾端如下所示： 因此，PolicyService似乎成功连接到message Broker。 Turbine AMQP服务器的日志结束： 编辑：下面的异常是当我停止收听涡轮流时抛出的，而不是当我尝试用仪表板收听时抛出的。 我对Turbane-AMQP的依赖关系如下：