我有pyspark数据帧，我想写到卡夫卡主题。: string (nullable = true) 我尝试过的代码 df.selectExpr("to_json(struct(*)) AS value") \.format("kafka") \.option("topic", "test") \
.sa
浏览 96提问于2020-06-17得票数 2
1回答
在PySpark中，如何将Dataframe转换为普通字符串？我将PySpark与Kafka一起使用，而不是硬编码代理名称，而是在PySpark中参数化了Kafka broker名称。这些变量为带有字符串的Dataframe类型。错误：
只能将字符串(不是Dataf
浏览 0提问于2021-03-05得票数 0
1回答
我正在尝试用spark做一个结构化的流媒体应用程序，主要思想是从kafka源读取，处理输入，写回另一个主题。我已经成功地让spark读写了kafka，但是我的问题出在处理部分。我尝试过foreach函数来捕获每一行并在写回kafka之前对其进行处理，但是它总是只做foreach部分，而不会写回kafka。但是，如果我从writestream中删除foreach部分，它将继续写入，但现在我丢失了处理。.ap
浏览 37提问于2019-07-17得票数 0
回答已采纳
当将Kafka消息转换为dataframe时，在将包作为参数传递时会出现错误。from pyspark.sql import SparkSession, Rowfrom kafka importKafkaConsumer
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages o
浏览 1提问于2020-06-12得票数 2
回答已采纳
我正在尝试将数据帧加载到Kafka主题。我在选择键和值时遇到错误。任何建议都会很有帮助。下面是我的代码， data = spark.sql('select * from job')
.writeStream.outputMode(outputMode='Append').format('ka
浏览 77提问于2020-06-14得票数 0
我编写了一些连接到kafka broker的示例代码，从主题读取数据并将其放入snappydata表。from pyspark.conf import SparkConffrom pyspark.sql import SQLContext, Row, SparkSessionfrom pyspark.rdd im
浏览 1提问于2020-11-30得票数 0
回答已采纳
1回答
有没有人尝试使用spark-steaming(pyspark)作为CDH中kerberos KAFKA的消费者？是不是CDH不支持？
浏览 2提问于2018-02-08得票数 1
当将Kafka消息转换为dataframe时，在将包作为参数传递时会出现错误。from pyspark.sql import SparkSession, Rowfrom kafka importKafkaConsumer
os.environ['PYSPARK_SUBMIT_ARGS'] = '--jars spark
浏览 5提问于2020-06-12得票数 1
2回答
我正在尝试读取通过Kafka和SparkStreaming获取的HDFS存储的数据。我使用的是一个Java应用程序，它使用JavaRDD.saveAsTextFile方法将一些任意数据保存到Hadoop。然后，在pyspark应用程序中，我尝试使用sparkContext.textFile读取数据。
问题是我读取的数据(使用python或localhost:50070中的“手工”)也包含元数据。
浏览 0提问于2018-01-22得票数 0
回答已采纳
我有一个代码(kafka_producer.py)，从csv >>中读取，创建Pandas dataframe >>，将熊猫的数据转换为星星之火>>调用>>方法，将消息发送给kafka。代码如下： kafkaBrokers='kafka.broker:9093'
caRootLocation='/path&
浏览 37提问于2022-05-27得票数 0
回答已采纳
我使用pyspark流从kafka获取数据，结果是一个dataframe，当我将dataframe转换为rdd时，它出错了： Traceback (most recent call last):File "/home/softs/spark-2.4.3-bin-hadoop2.6/py
浏览 56提问于2020-01-06得票数 3
我正在使用pyspark将Kafka主题中的数据作为流数据读取，如下所示： .appName("Spark Structured Streamingfrom Kafka") \ .readStream \ .option("kafka.bootstrap.ser
浏览 2提问于2020-03-12得票数 0
回答已采纳
我们有10个节点的AWS EMR集群，emr 5.5.0版本，Spark 2.1.0我们希望将此摘要数据(PySpark DataFrame)写入couchbase数据库。Couchbase Spark Connector支持PySpark吗？如果是，请分享一下如何使用
浏览 16提问于2017-08-05得票数 0
我想把流数据从kafka topic写到hive table。 global table data.select(col("key"),col("value&q
浏览 10提问于2022-03-23得票数 0
回答已采纳
简单地说，有没有可能将DStream流式传输到Kafka主题？
我有火花流作业，它做了所有的数据处理，现在我想把数据推送到卡夫卡主题。在pyspark中可以做到这一点吗？
浏览 1提问于2018-06-08得票数 0
我正在从kafka流式传输json数据，我需要转换为pyspark中的Dataframe。为了流媒体，我使用了下面的代码。from pyspark.streaming import StreamingContextfrom pyspark.sql,rdd): nf.toDF().show()
# C
浏览 3提问于2017-07-30得票数 0
从一个简单的Python脚本中，我每5秒将一些虚拟数据流到Kafka主题中。然后，使用Spark结构化流，我将这个数据流(每次一行)读入PySpark DataFrame中，其中包含startingOffset = latest。最后，我试图将这一行附加到已经存在的Cassandra表中。有一行数据正在成功地写入Cassandra表中，但我的问题是，每次覆盖时，都会被覆盖，而不是将追加到表的末尾。/bin/
浏览 1提问于2020-10-21得票数 0
回答已采纳
1回答
火花版本: 2.3.0卡夫卡版本: 2.2.0
我有一个卡夫卡生产者，它以avro格式发送嵌套数据，我正在尝试用火花流/结构化流写代码，这将反序列化从kafka到dataframe的avro，做转换，把它以拼花格式写成我能够在spark/scala中找到avro转换器，但是还没有增加对pyspark的支持。如何在火花放电中转换相同的内容。谢谢。
浏览 3提问于2019-05-02得票数 3
回答已采纳
我有下面的DataFrame，它有模式和data.Now，我只是将这些数据写入Kafka主题，这是vehicleData。vehicleData") \ Fi
浏览 1提问于2020-03-11得票数 0
1回答
我正在使用PySpark的DataFrame部分来分析来自Apache Kafka的数据。我遇到一些麻烦，需要一些帮助。from pyspark.sql import functions
# selected_df is dataframe come from kafka use spark.readStream.format("kafka")...windowed_group_1 = selected_df.withWatermark(&
浏览 21提问于2019-03-14得票数 1
云服务器
ICP备案
对象存储
云点播
实时音视频
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号