在Pyspark中使用流api读取Kafka主题-问题无法写入控制台或发送到任何其他接收器

文章/答案/技术大牛

发布

1回答

Spring Kafka JDBC连接器兼容性

spring-boot、spring-kafka、apache-kafka-connect、confluent-platform

Kafka JDBC connect兼容Spring-Kafka库吗？我确实关注了，但仍然有一些困惑。

浏览 4提问于2020-07-04得票数 0

2回答

python、apache-spark、pyspark、apache-kafka、spark-streaming

我有点卡住了，我在pyspark中使用spark streaming API阅读了一个Kafka主题，然后试图将它推到一个水槽中，无论是控制台还是另一个Kafka主题，我不确定我做错了什么，但整个过程有点卡住了我检查了有关于该主题的消息等。如果我使用基于java的消费者，我可以阅读并继续，但不知何故，对于pyspark无法消费和输出消息。我也把代

浏览 136提问于2021-07-02得票数 1

1回答

一个连接器的多个转换器

apache-kafka、apache-kafka-connect

卡夫卡连接器中有可能为单个接收器连接器提供多个转换器吗？例如，我们有一个用例，在该用例中，我们以格式A接收来自主题的数据，但希望将其流到两个位置，一个是格式A，另一个是格式B。在一个连接器中可以做到这一点吗？

浏览 1提问于2017-10-12得票数 0

回答已采纳

2回答

如何将数据从Google PubSub主题流到PySpark ( Google上)

python、pyspark、streaming、google-cloud-pubsub

我在Google PubSub中有一个主题的数据流。我可以看到使用简单Python代码的数据：def callback(message): print(datetime.now().strftime("%Y-%m-%d %H:%M:%S(使用订阅服务器subscriber_name)接收数据，并按照预期将其写入终端。我希望将来自主题的相同数据流到PySpark (RDD或dataframe)中</e

浏览 1提问于2018-09-17得票数 1

1回答

Pyspark结构化流处理

apache-spark、pyspark、spark-structured-streaming

我正在尝试用spark做一个结构化的流媒体应用程序，主要思想是从kafka源读取，处理输入，写回另一个主题。我已经成功地让spark读写了kafka，但是我的问题出在处理部分。我尝试过foreach函数来捕获每一行并在写回kafka之前对其进行处理，但是它总是只做foreach部分，而不会写回kafka。但是，如果我从writestream中删除foreach部分，它将继续写入，但现在我丢失了处理。如果有人能给我举一个例子，告诉我如

浏览 37提问于2019-07-17得票数 0

回答已采纳

1回答

卡夫卡连接与溪流对决

apache-kafka、apache-kafka-streams、apache-kafka-connect

我试着理解Connect给你买了什么流不给你的东西。我们有一个应用程序的一部分，我们想要使用一个主题并写到mariadb。为什么这是个坏主意？JDBC连接器给你买了什么？

浏览 0提问于2019-01-18得票数 11

回答已采纳

1回答

阅读单个kafka主题并插入多个甲骨文表

apache-kafka、apache-kafka-connect

我使用kafka使用mongo源连接器将多个集合数据发送到一个主题中，并使用jbc接收器连接器将数据插入到不同的oracle表中。在mongo源连接器中，我们为所有记录附加相应的集合名称，以便根据集合名称在接收器端处理信息。使用jdbc接收器连接器可以吗？我们是否可以通过节点. .js/ spring作为使用者应用程

浏览 2提问于2022-04-04得票数 0

1回答

用火花流读取脚本的输出输出

python、apache-spark、pyspark、spark-structured-streaming

我想用星火流api“捕获”这些值，将它们存储在拼花文件中，并应用一些推理代码，这是HiveQL编写的。我不是Scala人:/因此，如果可能的话，我更愿意在PySpark中找到一个解决方案，但我对任何建议都很满意。我知道可以读取来自Kafka流的数据流--例如，是否有类似的方法来读取发送到stdout的数据或连续写入文件的数据？提前感谢

浏览 0提问于2019-08-07得票数 3

回答已采纳

2回答

我应该使用什么: Kafka流或Kafka消费api或Kafka* connect*

apache-kafka、kafka-consumer-api、apache-kafka-streams、apache-kafka-connect

我想知道什么是对我最好的:卡夫卡流或卡夫卡消费api或卡夫卡连接？我想从主题中读取数据，然后进行一些处理并写入数据库。因此，我已经编写了使用者，但我觉得我可以编写Kafka流应用程序，并使用它的有状态处理器执行任何更改并将其写入数据库，这可以消除我的使用者代码，只需要编写db代码。我想插入我的记录的数据库是: HDFS -(插入原始JSON)

浏览 3提问于2017-09-04得票数 7

3回答

避免多次流查询

apache-spark、spark-structured-streaming

我有一个结构化的流查询，它深入到Kafka。此查询具有复杂的聚合逻辑。是否有一种方法可以将结构化流查询的结

浏览 1提问于2018-02-13得票数 5

回答已采纳

1回答

Spring @StreamListener条件不推荐替代方案

java、spring-boot、apache-kafka-streams、spring-cloud-stream、spring-cloud-stream-binder-kafka

我们让多个应用程序使用者收听相同的kafka主题，并且生产者在向主题发送消息时设置消息头，以便特定实例能够评估标头并处理消息。TestObj message) {} 在SpringCloudStream3.0.0中，@StreamListener被废弃了，我无法在</

浏览 11提问于2021-09-16得票数 5

1回答

Spark streaming kafka在google云平台中的集成问题

apache-spark、apache-kafka、google-cloud-platform、spark-streaming

我们在Google云平台上使用Bitnami Kafka 0.8.2 + spark 1.5.2。我们的spark streaming作业(消费者)没有接收到发送到特定主题的所有消息。它会收到50条消息中的1条(已在作业流中添加日志并已标识)。我们在kafka日志中没有看到任何错误。无法从kafka层进一步调试。控制台使

浏览 1提问于2015-12-24得票数 0

3回答

如何在星火流应用程序中使用Kafka主题？

apache-spark、pyspark、apache-kafka、spark-streaming

当我从Kafka主题创建一个流并打印它的内容时 os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell' from pyspark<

浏览 2提问于2019-12-08得票数 1

回答已采纳

2回答

如何配置无源卡夫卡通道？

apache-kafka、channel、flume

如果配置中没有指定源，则会发生故障。据医生说：有了水槽源和水槽--它为事件提供了一个可靠和高度可用的通道。使用Flume源代码和拦截器，但没有接收器--它允许将Flume事件写入Kafka主题，供其他应用程序使用。使用水槽接收器，但没有源，这是一种低延迟、容错的方式，可以将事件从Kafka发送到Flume接收器</em

浏览 0提问于2019-04-10得票数 0

1回答

火花结构化流确认消息

scala、apache-spark、apache-kafka、spark-structured-streaming

我使用从卡夫卡主题(例如topic1)中读取，并使用接收器写入另一个主题(主题1-结果)。我可以看到，在使用Sink编写到另一个主题之后，这些消息并没有从Topic1中删除。// Subscribe to 1 topic .readStream .format("kafka&qu

浏览 2提问于2020-02-11得票数 0

回答已采纳

1回答

“精确一次”是否只适用于流(topic1 -> app -> topic2)？

java、apache-kafka、apache-kafka-streams

它将这些序列化的消息发送到Kafka主题。App2监听该主题，将消息调整为不同的格式，并通过HTTP将这些经过调整的消息发送到各自的目的地。看起来，Kafka流api是针对单个应用程序的，该应用程序从Kafka主题获取输入，进行处理，并将其输出到另一个Kafka主题，这似乎不适用于我们对Kafka的使用。以下是的摘录现在，流

浏览 0提问于2019-05-09得票数 2

回答已采纳

2回答

如何将消息从kafka接收器路由到多个主题

apache-kafka、spring-integration、spring-xd、kafka-producer-api

我有一个spring处理器模块，带有errorChannel和outputChannel的http出站网关.使用HTTP200的任何消息都会发送到outputChannel，其余消息都在failureChannel中。现在，http处理器模块用TopicX的kafka出站适配器连接到Kafka-Sink .TopicX只接收HTTP200个消息以供进一步处理。现在，我们需要将failureChannel中的消息路由到TopicY。如何在卡夫卡水槽中发送多个卡夫卡<em

浏览 5提问于2017-03-28得票数 2

回答已采纳

1回答

Kafka中的动态流拓扑

java、apache-kafka、apache-kafka-streams

使用Kafka流DSL 创建Kafka流时例如:当我们开始的时候，我们定义了一个拓扑来读取一个主题(来源)和一个目标主题(Sink)。但是，在进行配置更改时，我们现在需要从两个不同的主题(如果需要的话是两个源)进行读取，并将其写入单个目标主题。从我们现在构建的内容

浏览 0提问于2019-06-18得票数 1

回答已采纳

2回答

卡夫卡-> DataStream -> MongoDB

mongodb、hadoop、apache-kafka、apache-flink

我想设置Flink，这样它就可以将数据流从Apache转换并重定向到MongoDB。出于测试目的，我构建在flink流-Connectors.kafka示例()之上。卡夫卡流是正确红色的链接，我可以映射他们等，但问题发生时，我想保存每一个接收和转换的信息到MongoDB。我发现的关于MongoDB集成的唯一例子是来自github的flink-mongodb测试。不幸的是，它使用的是静态数据源(数据库)，而不是数据流。我认为应该有一些DataStream.addSink

浏览 3提问于2016-02-02得票数 11

回答已采纳

1回答

Spark流式数据以更新SQL Server (事务)

sql-server、spark-streaming

目前我有一些pyspark代码，它是从kafka主题读取数据(readStream)，我计划使用事务更新SQL Server表。流数据将具有所有三个插入、更新、删除事务。

浏览 2提问于2020-01-29得票数 0

点击加载更多

Spring Kafka JDBC连接器兼容性