《stream》专题
-
Kafka Streams 2.5.0需要输入主题
从Kafka Streams 2.5.0开始，拓扑似乎必须包含一个输入主题。在Kafka2.4.1（以及更早的版本）中，情况并非如此。 我有一个应用程序，其中的拓扑只是创建一些全局状态存储，从其他应用程序写入的主题中读取数据。 使用Kafka 2.5.0，我得到以下错误： 如果添加一个虚拟输入主题（例如，通过)，应用程序启动良好。 这种行为是意料之中的，还是Kafka Streams 2.5.0中
-
Kafka流：使用Spring Cloud Stream为每个主题集定义多个Kafka流
配置的application.yaml如下所示。这个想法是我有3个输入和3个输出主题。该组件从input topic获取输入，并将输出提供给OutputTopic。 引发的异常为 谁能帮助我与Kafka Streams Spring-Kafka代码样本处理与多个输入和输出主题。 更新：2021年1月21日
-
Spring Cloud StreamListener@Output KStream Serdes似乎不起作用
我有一个流监听器，作为 其中@input（“requesti”）配置如下所示； 发送记录ProducerRecord(TOPIC=Request，Partition=null,headers=recordheaders(headers=[recordheader(key=Key_TypeId，值=[106,97,118,97,46,117,116,105,108,85,85,85,73,68])，
-
Spring Cloud Stream Producer在使用Spring Kafka Consumer时添加“垃圾”字符
我们的微服务使用JDK8、Spring Cloud Stream API和Spring Boot 1.5.2版本，用于在Kafka队列上生成消息。对于消费者API，我们使用Spring Kafka 1.1.2.release，特别是。消息类型为应用程序json 谢谢
-
Redis streams：组/消费者命名和清理
我有一个相当原始的流用例：多个生产者和一个消费者周期性地批量处理消息。有时会有多个使用者，这就是为什么我想从“传统”的不可靠队列（rpush/lrange/ltrim）切换到流。 实际上有两个相关的问题： > 必须先通过创建消费者组，然后才能从中创建。每次一个新的消费者开始（使用随机名称）时，它都会被添加到消费者列表中，但即使在流程结束后，它仍会保留在那里。我的理解是，在这种情况下，我需要将所有挂
-
Spark Streaming：执行器与自定义接收器的数量
为什么Spark有一个工作节点和四个执行器，每个执行器有一个核心，却无法处理自定义接收器？？ 如果执行器在Spark Streaming中有一个单核，那么不通过自定义接收器处理传入数据的原因是什么？ 我正在独立模式下运行Spark。我在Spark流媒体应用程序的自定义接收器中获取数据。我的笔记本电脑有4个内核。 master="火花：//lappi： 7077" $spark_path/bin/s
-
Stream.count（）vs C有什么区别ollectors.counting（）
接口中的方法与中的方法之间有什么区别吗？一般应该使用哪一个？使用一个比另一个有任何性能优势吗？
-
如何使用Java streams[重复]查找不同对象的计数
我有一个Thing对象列表： <代码>列表 现在，我想使用Java streams或任何util函数来获得id的不同计数：我希望输出为， 我如何才能实现这一点？
-
获取错误NoClassDefFoundError:org。阿帕奇。火花内部的登录Kafka Spark Stream
我在运行时通过运行下面的java代码得到一个错误， 日志记录是否需要包含任何依赖项，例如log4js或其他东西？ 为什么编译时不出现这个错误，这样更容易... 这是我的Java代码， 依赖项使用， 获取以下错误，
-
SparkContext和StreamingContext可以在同一个程序中共存吗？
我正在尝试设置一个Spark流代码，该代码从Kafka服务器读取行，但使用另一个本地文件中编写的规则来处理它。我正在为流式数据创建StreingContext，并为其他应用所有其他火花功能的人创建sparkContext——例如字符串操作、读取本地文件等 但这会引发以下错误
-
通过spark streaming或flume将Xml转换为Avro，从Kafka转换为hdfs
我想将xml文件转换为avro。数据将采用xml格式，并将首先触及Kafka主题。然后，我可以使用flume或spark streaming来摄取xml并将其转换为avro，然后将文件放在hdfs中。我有一个cloudera环境。 当avro文件到达hdfs时，我希望能够稍后将它们读入hive表。 我想知道做这件事最好的方法是什么？我尝试过自动模式转换，比如spark avro（这没有spark流
-
将RDD转换为DataFrame Spark Streaming时的ClassCastException
嗨，伙计们，我有下一个问题。我正在使用Java的Apache Spark Streaming v1.6.0来获取来自IBMMQ的一些消息。我为MQ制作了自定义接收器，但我遇到的问题是我需要将RDD从JavaDStream转换为DataFrame。为此，我使用foreachRDD迭代JavaDStream，并定义了DataFrame的模式，但当我运行作业时，第一条消息会引发下一个异常： Java语言
-
Spark Streaming异常：java.util.NoSuchElementException：None.get
我正在将数据转换为数据帧，将其写入HDFS： 密码 已创建文件夹，但未写入文件。 程序因以下错误而终止： 在我的pom中，我使用了各自的依赖项： 火花-core_2.11 火花-sql_2.11 火花-streaming_2.11 火花流-kafka-0-10_2.11
-
stream.max（）如何处理相等性？
虽然我怀疑答案是“没有具体说明”... 如果中有多个“最大/最低”元素，传递给max或min方法的认为这些元素相等（返回)，是否指定了将找到哪个元素？
-
Guava的Streams：：FindLast实现
我正在研究guava的的实现，在试图理解它时，有几件事我根本无法理解。下面是它的实现： 老实说，实际上实现并不那么复杂，但我发现有些事情有点奇怪（如果这个问题被归结为“基于意见”，我会承担责任，我明白可能会发生这种情况）。 首先是创建类，可以用单个元素的数组替换它： 并且使用简单到： 然后整个方法可以分成3个部分：