《stream》专题

我希望火花流的输出在翻转窗口的末端发送到水槽，而不是在批处理间隔。 我从一个Kafka流读取并输出到另一个Kafka流。 查询和写入输出的代码如下： 当我在一分钟的窗口内为一个特定用户发送多个记录时，我希望在一分钟结束时这些事件的总数。 但我在输出Kafka流上获得了多个输出，并在其中写入了间歇聚合。 如。 我将在一分钟内发送以下7条记录，但间隔一定时间。 我得到的结果是： 可以看到，输出在同一个

我是Spark结构化流处理的新手，目前正在处理一个用例，其中结构化流应用程序将从Azure IoT中心-事件中心（例如每20秒）获取事件。 任务是使用这些事件并实时处理。为此，我在下面用Spark Java编写了Spark结构化流媒体程序。 以下是要点 目前我已经应用了10分钟间隔和5分钟滑动窗口的窗口操作。 水印被应用在以10分钟间隔的eventDate参数上。 目前，我没有执行任何其他操作，只

回顾Java 8API设计时，我对参数的泛型不变性感到惊讶： 同一个API的一个看起来更通用的版本可能已经在对的单独引用上应用了协方差/逆方差，例如： 这将允许当前不可能的以下操作： 解决办法是使用方法引用将类型“强制”为目标类型： C#没有这个特殊的问题，因为定义如下，使用声明-站点方差，这意味着任何使用的API都可以免费获得这个行为： 与建议的设计相比，现有的设计有什么优势（以及可能的，决策的

因此，我们正在寻找防止这种融合发生的方法，因此数据流将窗口与窗口的后处理分开。这样，我们期望Dataflow能够再次分配多个工作人员来进行激发窗口的后处理。 到目前为止我们所尝试的： null 最后两个操作确实创建了第三个集群操作(1/processing2/windowing3/post-processing)，但我们注意到，在开窗之后，仍然是同一个worker在执行所有操作。 是否有任何解决方

查找缓存的引用表位于BigQuery中，我们可以读取它并将其作为ParDo操作的侧输入传入，但无论我们如何设置触发器/窗口，它都不会刷新。 根据这里的I/O页面（https://beam.apache.org/documentation/io/build-in/)，它说Python SDK只支持BigQuery接收器的流，这是否意味着BQ读取是一个有界源，因此不能在此方法中刷新？ 试图在源上设置非

我想看看我是否可以在docker容器中docker compose的帮助下连接Spring Cloud Stream Kafka，但我被卡住了，我还没有找到解决方案，请帮助我。 我正在使用Spring微服务。到现在也没找到任何帮助。 Docker-compose与Kafka和Zookeeper： Docker-Compose与我的Spring服务： App.properties我的服务： 来自do

此操作的行为显式不确定。对于并行流管道，此操作不能保证尊重流的相遇顺序，因为这样做会牺牲并行性的好处。对于任何给定的元素，操作可以在库选择的任何时间和线程中执行。 我知道如果代码使用而不是，就不能保证它能够工作，但是由于它使用的是顺序流（Javadoc没有说明这一点），所以我不确定。这是否保证始终有效，或者代码需要使用而不是才能有效？ 编辑：我认为这个问题不是Java8 Stream中forEac

我有一个人的HashSet。一个人有一个名字，最后一个名字和年龄，比如：人（“汉斯”，“男人”，36） 我的任务是得到一份17岁以上的人的名单，按年龄对他们进行排序，并用姓氏来表示他们的名字，如：[“Hans Man”，“另一个名字”，“另一个名字”] 我只允许导入： 我的想法是先对它们进行排序，将名称映射到不同的流中，然后对它们进行压缩，但这不起作用。 提前谢谢你

我需要将偏移量重置为一个数字。 详细要求：我的应用程序正在使用来自 kafka 主题的消息并将其转储到 DB 中，假设 DB 在处理已使用的消息时出现故障（offset=10），直到数据库关闭时应用程序使用的消息，直到偏移量 20。 现在DB在处理第20个偏移量消息时再次出现，现在我想再次将偏移量重置为10，以便将数据保存在数据库中。 我可以通过编程（Spring启动）来实现吗？我正在使用Spri

在scala.collection.traversablelike$$anonfun$map$1.apply(traversablelike.scala:244) 在scala.collection.indexedSeqOptimized$class.foreach(indexedSeqOptimized.scala:33) 在scala.collection.mutable.arrayops$o

这是否意味着我的整个应用程序只能连接到单个Kafka集群，或者KafkaStreams的每个实例只能连接到单个集群？ 我可以创建多个连接到不同集群的具有不同属性的KafkaStreams实例吗？

我想有一个客户端应用程序与请求/响应语义学调用另一个应用程序，这是一个Kafka流应用程序。 我的客户端应用程序基于此示例（基本上没有变化）。我需要从客户端接收消息的应用程序是Kafka Streams应用程序。但是包含相关id的消息头丢失。 Kafka Streams应用程序是一个简单的拓扑结构，用于测试此。。。 对于这个POC，我保持它的简单性，让客户机和服务器“同意”主题名称（和）。所以在这

例如，我有一个类包含name和surname字段。 我想从的中收集一个的（名字和姓氏都在一起），但似乎无法对每个列表使用两次map或无法对每个列表使用两次stream。 我做错了什么？

源代码：