PySpark :将Spark Dataframe写入Kafka主题

文章/答案/技术大牛

发布

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-spark、pyspark、spark-structured-streaming

我正在尝试将数据帧加载到Kafka主题。我在选择键和值时遇到错误。任何建议都会很有帮助。下面是我的代码， data = spark.sql('select * from job') .writeStream.outputMode(outputMode='Append').f

浏览 77提问于2020-06-14得票数 0

1回答

CDH星火消费者kerberos kafka

pyspark、apache-kafka、spark-streaming、kerberos、cloudera-cdh

有没有人尝试使用spark-steaming(pyspark)作为CDH中kerberos KAFKA的消费者？是不是CDH不支持？

浏览 2提问于2018-02-08得票数 1

1回答

将星火流PySpark数据格式写入Cassandra覆盖表而不是追加

apache-spark、apache-kafka、cassandra、spark-structured-streaming、spark-cassandra-connector

从一个简单的Python脚本中，我每5秒将一些虚拟数据流到Kafka主题中。然后，使用Spark结构化流，我将这个数据流(每次一行)读入PySpark DataFrame中，其中包含startingOffset = latest。最后，我试图将这一行附加到已经存在的Cassandra表中。有一行数据正在成功地写入Cassandra表中，但我的问题是，每次覆盖时，都会被覆盖，而不是将追加到表

浏览 1提问于2020-10-21得票数 0

回答已采纳

1回答

基于dashbord可见性的PySpark结构化流查询

apache-spark、pyspark、snappydata

我编写了一些连接到kafka broker的示例代码，从主题读取数据并将其放入snappydata表。, Row, SparkSessionfrom pyspark.rdd import RDDfrom pyspark.sql.functions import col, e

浏览 1提问于2020-11-30得票数 0

回答已采纳

1回答

writeStream:在一个单独的json文件中的每个数据帧行

pyspark、spark-streaming、pyspark-dataframes

我正在使用pyspark将Kafka主题中的数据作为流数据读取，如下所示： .appName("Spark Structured Streamingfrom Kafka") \ .readStream \ .optio

浏览 2提问于2020-03-12得票数 0

回答已采纳

1回答

Spark Structred Streaming Pyspark Sink Csv不附加

python-3.x、apache-spark、pyspark、apache-kafka、spark-structured-streaming

向Kafka topic写入json，并从kafka Topic读取json。实际上，我是逐行订阅主题和编写控制台的。但我必须接收/写入文件csv。但是我不能。我只写了一次csv，但没有附加。import pysparkfrom pyspark.sql.types import * import pyspark.sql.functionsas fun

浏览 23提问于2020-01-11得票数 1

回答已采纳

1回答

如果Maven坐标必须是'groupId:artifactId:version‘PySpark和Kafka格式的

apache-spark、pyspark、apache-kafka

当将Kafka消息转换为dataframe时，在将包作为参数传递时会出现错误。from pyspark.sql import SparkSession, Rowfrom kafka importKafkaConsumer os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages o

浏览 1提问于2020-06-12得票数 2

回答已采纳

1回答

如何以编程方式将Kafka主题加载和流到PySpark数据

python、pyspark、apache-kafka、kafka-consumer-api

我试着阅读来自卡夫卡主题的信息，并从中创建一个数据框架。可以从主题中提取消息，但无法将其转换为数据文件。任何建议都会有帮助。import pysparkfrom pyspark.context import SparkContextspark = Spark

浏览 1提问于2020-06-12得票数 1

2回答

如何显示一个流DataFrame (AnalysisException显示失败)？

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

所以我在Kafka主题中有一些数据，我把这些流数据放到一个DataFrame中。我想要显示DataFrame中的数据：from kafka import KafkaProducer from pyspark.sql import SparkSession, DataFrameorg.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0,org.apache.<

浏览 3提问于2017-07-14得票数 5

回答已采纳

1回答

Pyspark Kafka结构化流:写出时出错

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

我能够在PySpark中通过两个不同的步骤从Kafka主题读取流，并将(转换后的)数据写回另一个Kafka主题。执行此操作的代码如下所示：df = spark \ .format("kafka") \ .outputMode("append") \

浏览 0提问于2018-10-07得票数 0

2回答

PySpark Kafka py4j.protocol.Py4JJavaError:调用o28.load时发生错误

apache-spark、pyspark、apache-kafka

浏览 5提问于2020-06-12得票数 1

1回答

由于文件格式不匹配，PySpark数据不能保存在Hive中

apache-spark、pyspark、apache-spark-sql、spark-streaming

我想把流数据从kafka topic写到hive table。 .format("kafka") \ .options(**kaf

浏览 10提问于2022-03-23得票数 0

回答已采纳

1回答

Kafka和pyspark程序:无法确定dataframe为空的原因

apache-spark、pyspark、apache-kafka、spark-structured-streaming、spark-kafka-integration

下面是我使用kafka和pyspark编写的第一个程序。代码似乎运行无异常，但我的查询的输出是空的。我该如何解析？代码：from pyspark.sql import SQLCont

浏览 11提问于2021-09-12得票数 0

1回答

无法使用df.show()在控制台上打印数据

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

我正在使用readStream阅读kafka主题，并尝试在控制台上显示数据。df = spark.readStream.format("kafka")\.option("kafka.bootstrap.servers", kafkaServer)\.option("subscribeFile "C:/spark/spark-3.2.1-bin-hadoop2.7\python\pys

浏览 5提问于2022-03-30得票数 0

1回答

如何优化卡夫卡主题的结构化流消费分区策略？

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

我对kafka非常陌生，我试图将数据写到一个主题中，并从同一个主题中读取数据(我们现在作为一个源团队来摄取数据)。因此，我们在对Kafk主题进行写操作，并从相同的主题中使用)。我在星火壳上编写了下面的代码，以便将数据写入一个Kafka主题。pyspark --packages io.delta:delta-core_2.11:0.6.1,org.apache.spark:s

浏览 3提问于2021-09-08得票数 1

回答已采纳

1回答

从多列DataFrame中火花批写卡夫卡主题

apache-spark、apache-kafka、apache-spark-sql

批处理之后，我需要向Kafka主题写入包含多个不同列的结果DataFrame。根据下面的火花文档，写入Kafka的Dataframe在模式中应该有以下强制列：正如我前面提到的，我有更多的列有值，所以我有一个问题--如何正确地将整个DataFrame行作为一条消息从我的Spark应用程序发送到Kafka主题？我是否需要用一个值列(包含联接值)<em

浏览 0提问于2018-11-23得票数 0

回答已采纳

1回答

将Dataframe激发到StringType

json、apache-spark、pyspark、apache-kafka

在PySpark中，如何将Dataframe转换为普通字符串？我将PySpark与Kafka一起使用，而不是硬编码代理名称，而是在PySpark中参数化了Kafka broker名称。Json文件保存了代理详细信息，Spark读取这个Json输入并将值赋值给变量。这些变量为带有字符串的Dataframe类型。当我将dataframe传递给Pyspark连

浏览 0提问于2021-03-05得票数 0

1回答

从Kafka主题失败将数据写入Cassandra表

apache-spark、pyspark、cassandra、spark-cassandra-connector

我写了简单的工作，它消耗了卡夫卡主题的数据，并将其写入卡桑德拉表。Cassandra作业失败时-我使用的是Kafka3.2.1，Cassandra 3.11.13，pyspark 3.3.0。我的火花-服从命令- spark-submit --packages org.apache.spark:spark<

浏览 6提问于2022-08-25得票数 1

回答已采纳

4回答

如何在结构化流式传输中将数据帧转换为rdds？

apache-spark、spark-streaming

我使用pyspark流从kafka获取数据，结果是一个dataframe，当我将dataframe转换为rdd时，它出错了： Traceback (most recent call last):File

浏览 56提问于2020-01-06得票数 3

点击加载更多

Pyspark结构化流处理