1回答
在将rdd转换为dataframe时使用mapPartitions的一个任务
python、apache-spark、pyspark、apache-spark-sql
我感到困惑的是，为什么在将得到的rdd.mapPartitions转换为DataFrame时，Spark似乎使用了1任务。)，将非SQL函数应用于数据块(mapPartitions on RDD)，然后转换回DataFrame，以便我可以使用DataFrame.write进程。我可以从DataFrame -> mapPartitions开始，然后使用像saveAsTextFi
浏览 4提问于2016-11-22得票数 17
回答已采纳
3回答
pySpark将mapPartitions的结果转换为spark DataFrame
python、apache-spark、pyspark
我有一个作业需要在分区的spark数据帧上运行，该进程如下所示： rdd = sp_df.repartition(n_partitions, partition_key).rdd.mapPartitions(lambda x: some_function(x)) 结果是pandas.dataframe的rdd， type(rdd) => pyspark.rdd.PipelinedRDD
type(rdd.collect()[0]) => pandas.cor
浏览 124提问于2019-12-10得票数 4
回答已采纳
1回答
Pyspark体验
apache-spark、pyspark、user-defined-functions
我意识到UDF有时会减慢代码的速度。我想知道你的经历。您应用了什么UDF功能(不能仅用Pyspark代码实现)。有什么有用的UDF函数可以帮助您清理数据吗？除了Pyspark文档之外，还有什么资源可以帮助我学习UDF函数吗？
浏览 8提问于2022-03-03得票数 0
1回答
PySpark:关于调用df.foreach方法的PicklingError
python、apache-spark、pyspark、pickle
我有一个代码(kafka_producer.py)，从csv >>中读取，创建Pandas dataframe >>，将熊猫的数据转换为星星之火>>调用>>方法，将消息发送给kafka。): spark = get_spark_session("session_
浏览 37提问于2022-05-27得票数 0
回答已采纳
1回答
如果mapPartitions与toDF()一起使用，那么为什么很少有分区被处理两次？
apache-spark、pyspark
当使用RDDs时，使用mapPartitions很好。在本例中，当使用rdd.mapPartitions(mapper).collect()时，所有工作都按预期进行。但是，当转换为DataFrame时，一个分区将被处理两次。下面是下一个简单示例的输出。当只有两个分区时，我们可以3次读取函数的执行方式。令人感到奇怪的是，其中一次执行被忽略了，正如我们在DataDrame结果中所看到的那样。：
fro
浏览 3提问于2020-10-02得票数 1
回答已采纳
1回答
可能将数据分解为拓扑的各个部分
python、pandas、apache-spark
我有一千万的记录数据。我的要求是，我需要对熊猫的这些数据做一些操作，而且我对所有1000万条记录都没有记忆。p_df = chunked_df.toPandas()如何通过记录计数将我的数据分割成相等的x-部分或部分，比如一次一百万。这两种解决方案都是可以接受的，我只需要用更小的块来处理它。
浏览 1提问于2018-10-26得票数 10
回答已采纳
1回答
如何将来自RDD.mapPartitions()的Pandas数据帧转换为Spark DataFrame？
python、pandas、apache-spark、pyspark
我有一个Python函数，它返回一个Pandas DataFrame。我在Spark2.2.0中使用pyspark的调用这个函数。但是我不能将mapPartitions()返回的RDD转换为Spark DataFrame。说明问题的简单代码： pdf = pd.DataFrame(list(data), columns
浏览 11提问于2018-01-31得票数 3
回答已采纳
4回答
如何在保留现有模式的同时从行中创建DataFrame？
python、pandas、apache-spark、pyspark、pyspark-sql
如果我调用map或mapPartition，并且我的函数接收来自PySpark的行，那么创建本地PySpark或Pandas DataFrame的自然方法是什么？合并行并保留架构的东西？目前，我所做的事情如下： rows = [x for x in partition]
dfpart = pd.DataFrame(rows
浏览 0提问于2015-12-23得票数 2
回答已采纳
1回答
使用PySpark和paho向MQTT代理发送foreachRDD结果
apache-spark、pyspark
我试图向MQTT代理发送一个带有计算结果的DStream，但是foreachRDD一直在崩溃。
我正在运行Spark2.4.3与Bahir的MQTT订阅，从git主编译。到目前为止一切都很顺利。在尝试用MQTT发布我的结果之前，我尝试了saveAsFiles()，这起作用了(但不是我想要的)。/python/lib/pyspark.zip/pyspark/rdd.py", l
浏览 0提问于2019-07-15得票数 1
回答已采纳
1回答
Spark Structured Streaming :支持mapPartitions吗？
apache-spark、pyspark
是否可以在Spark Structured streaming中使用mapPartitions？遇到这些错误dataframe_python.mapPartitions(processfunction)选项2：‘带有流源的查询必须使用writeStr
浏览 15提问于2019-08-01得票数 1
1回答
在同一个DataFrame中的Pyspark嵌套循环。如何迭代？
loops、dataframe、join、pyspark、nested
TL;DR: --我正试图在pyspark中实现一个嵌套循环。正如您可能看到的，我希望嵌套循环在每次迭代中从下一行开始(相对于第一个循环)，以减少不相关的迭代。, StructField('class', StringType(), True),rdd = spark.sparkContext.parallelize(people)
df = spark.creat
浏览 2提问于2020-03-28得票数 0
回答已采纳
1回答
测量时间火花操作(装载、处理、写入)
mongodb、pyspark
我在PySpark中编写了一些代码，将一些数据从MongoDB加载到，应用一些过滤器，处理数据(使用RDD)，然后将结果写回MongoDB。load() #df_initial is a Spark dataframerdd_to_process = df_filtered.rdd
processed_rdd = rdd_to_process.mapPartitio
浏览 6提问于2017-09-21得票数 0
回答已采纳
1回答
Apache Spark如何支持不同语言的API
api、apache-spark
我一直想知道像Apache Spark这样支持不同语言API的系统的高级架构。例如，Spark支持Scala、Python和Java中的API。
尽管该软件是用Scala编写的，但它支持多语言API。有人能解释一下这样的系统的高层架构吗？
浏览 0提问于2016-03-13得票数 3
1回答
将自定义函数应用于数据type数组类型的列
python、pyspark、spark-dataframe、rdd
我有一个名为‘count’的列的dataframe，我想将一个自定义函数"do_something“应用于列的每个元素，即每个数组。我不想修改dataframe，我只想做一个单独的操作与列计数。在上面的行上，它失败了
/usr/hdp/2.5.3.0-37/spark/python/pyspark/rdd.py in fo
浏览 2提问于2017-09-22得票数 1
回答已采纳
5回答
来自自定义模块的函数不是在PySpark中工作，而是在交互模式下输入时工作。
pyspark、pyspark-sql
我已经编写了一个模块，其中包含了在PySpark DataFrames上工作的函数。它们对DataFrame中的列进行转换，然后返回一个新的DataFrame。下面是代码的一个示例，缩短为只包含其中一个函数：from pyspark.sql import types as t
importmyf (PySpark DataFra
浏览 6提问于2016-03-10得票数 7
回答已采纳
4回答
如何在结构化流式传输中将数据帧转换为rdds？
apache-spark、spark-streaming
我使用pyspark流从kafka获取数据，结果是一个dataframe，当我将dataframe转换为rdd时，它出错了： Traceback (most recent call last):
File-2.4.3-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/sql/dataframe.py", line 91, in
浏览 56提问于2020-01-06得票数 3
1回答
OverflowError: mktime参数超出范围
python、datetime、pyspark
most recent call last):
File "/Users/pm/opt/spark-3.3.0-bin-hadoop3/python/lib/pyspark.zip/pyspark/File "/Users/pm/opt/spark-3.3.0-bin-hadoop3/python/lib/pyspark.z
浏览 5提问于2022-11-28得票数 0
2回答
如何在Pyspark中使用Scala类
python、scala、apache-spark、pyspark、apache-spark-sql
我已经搜索了一段时间了，如果有任何方法在Pyspark中使用一个Pyspark类，我还没有找到任何关于这个主题的文档或指南。假设我在Scala中创建了一个简单的类，它使用了一些apache-spark库，如下所示： def exe(): DataFrame = {
import sqlContext.impli
浏览 3提问于2016-03-15得票数 29
回答已采纳
1回答
使用python火花向Kafka发送大型CSV
python、apache-spark、apache-kafka、pyspark、kafka-python
我正试着给卡夫卡送一辆大的CSV。基本结构是读取CSV的一行并使用标头将其压缩。a = dict(zip(header, line.split(",")message = json.dumps(a)producer = SimpleProducer(kafka)
producer.send_messages("topic", message
浏览 4提问于2015-08-31得票数 6
回答已采纳
2回答
在熊猫(python)中，能不能将星星之火(scala)转换成数据(Python)
pandas、scala、dataframe、apache-spark
Dataframe是使用scala创建的。val someDF = spark.createDataFrame( spark.sparkContext.parallelize(someData), StructType(someSchema) )PySpark提供了.toPandas()来将火花数据转换成熊猫，但是scala(我可以找到)并没有对应的数据。
浏览 1提问于2020-04-05得票数 0
回答已采纳
云服务器
ICP备案
对象存储
云点播
即时通信 IM
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号