《stream》专题

我想使用Kafka流处理器应用编程接口，并在预定的标点符号函数中每分钟生成一些消息。Kafka流能保证这些消息恰好一次写入输出主题吗？ 我知道在Kafka Streams中可以进行一次处理，因为它通过以下操作生成一个事务： 提交输入主题的偏移量 这个概念是否扩展到处理器API中的标点符号或函数，对于这些标点符号或函数，没有需要提交的关联输入消息？ 例如，此标点器函数迭代键值状态存储中的项。每个商品

我使用kafka流来消耗来自一个主题的JSON字符串，处理并生成存储在另一个主题中的响应。然而，需要对响应主题产生的消息需要采用avro格式。 我已经尝试使用键作为字符串序列和值作为规范AvroSerde 以下是我创建拓扑的代码： 以下是我的配置 当我尝试使用该示例时，我看到了以下错误：

对于第一个方案，我想将 Spark 结构化流式处理与入口中的 Iot Hub 和 Cosmos DB 用于输出。我使用了以下连接器： azure-eventhubs-spark_2.11-2.3.2 azure-cosmosdb-spark_2.3.0_2.11-1.2.0 （不直接与 Maven，通过 import uber jar 事实上，当我试图将数据导入Cosmos DB时，我收到以下错误

我刚刚复制了spark streaming wodcount python代码，并使用spark-submit在spark集群中运行wordcount python代码，但它显示了以下错误： 我确实构建了jar spark-streaming-kafka-assembly2.10-1.4.0-snapshot.jar。我使用以下脚本提交:bin/spark-submit/data/spark-1.

您能帮我找出更好的解决方案吗？下一种情况：Web元素属性以字符串形式收集，格式为： 目标：创建属性及其值的映射 。注意：属性中的值duplicates是可能的，例如“none”。

在Java8流中，允许我修改/更新内部的对象吗？为。:

请考虑以下类： 注意：很重要的一点是，我不能修改这个类，因为我是从外部API中使用它的。 还要考虑以下订单层次结构： 通过递归地使用（以及一个helper类），我已经设法做到了这一点，如下所示： 这是helper类： 以下一行： 产生以下输出： 到目前为止还不错。结果是绝对正确的。 但是，在阅读了这个问题之后，我对在递归方法中的用法有些担心。特别是，我想知道流是如何被扩展的（如果这是术语的话）。因

例如，有两个列表： 使用Stream，我想创建一个由这些列表组成的映射，其中清单1是键，清单2是值。要做到这一点，我需要创建一个辅助列表： list0按List1::Get和List2::Get的顺序使用。有没有一种不创建list0的更简单的方法？我尝试了以下代码，但没有起作用：

我有，我只想有一个。我尝试使用，但它不起作用，因为它无法将SortedSet的流扁平化。

我使用方法处理消息。 我已经找到了这个任务，它仍然是打开的。https://issues.apache.org/jira/browse/kafka-5632？src=confmacro

所以我有一个场景，我的云流处理器函数从一个Kafka topic-1读取消息，并将消息生成到另一个Kafka Topic-2。但是这个过程必须及时运行，比如函数应该等待5分钟，然后它应该启动（消耗n生产）1分钟，然后1分钟后再次等待5分钟。有人能帮我做这件事吗？

在Kafka Stream DSL Join中，是否有方法从部分传入或访问消息键？ 但是，传入的和记录不能访问kafka消息键，因为这是一个单独的字符串。他们拥有的唯一内容是信息本身，而不是密钥。 是否有一种方法可以从连接lambda内部访问密钥？我可以做的一件事是简单地将消息键添加为消息本身的一部分，并在那里作为常规字段访问它，但我想知道框架是否提供了直接访问它的方法？

我如何使用java流以相同的性能过滤一个对象？ HashSet包含： 现在，我要筛选来生成一个新的，它具有唯一的-属性。 所有具有相同名称的对象都应被视为平等的对象： 我可以通过创建一个新的列表并使用2个for循环来解决这个问题，方法是索引下一个元素并查找（如果它已经放入列表中）。然而，也许有一个班轮这样的问题？可能使用？

下面是它的名称 这是配对类 更新： 测试了斯图尔特下面的功能，它似乎工作很好。下面的操作区分每个字符串的第一个字母。我想要弄清楚的是，ConcurrentHashMap如何只为整个流维护一个实例 输出为...

我只需要阅读下面的json文件。从该文件中读取完整的，然后将其流式传输。 上面的代码不是在读取json文件，而是在打印 你能给我提供样品参考吗？