对pyspark dataframe执行重复数据删除时遇到内存错误
对于pyspark dataframe执行重复数据删除时遇到内存错误的问题，可以采取以下几种解决方法：
- 增加内存资源：可以尝试增加集群的内存资源，以提供更多的内存空间来处理数据删除操作。可以通过增加集群的节点数或者增加每个节点的内存大小来实现。
- 分批处理数据：如果数据量较大，可以考虑将数据分批处理，每次处理一部分数据，避免一次性加载全部数据导致内存错误。可以使用pyspark的分区功能，将数据分成多个分区，然后逐个分区进行数据删除操作。
- 优化代码逻辑：检查代码逻辑是否存在不必要的计算或者重复计算，尽量减少内存占用。可以使用pyspark提供的一些优化技巧，如使用合适的数据结构、避免不必要的转换操作等。
- 增加数据处理节点：如果集群中只有一个节点在处理数据删除操作，可以考虑增加更多的节点来并行处理数据，以提高处理速度和减少内存占用。
- 使用持久化存储：如果内存资源有限，可以考虑将数据持久化到磁盘上，以释放内存空间。可以使用pyspark的持久化功能，将数据缓存到内存或者磁盘上，以便后续的操作使用。
- 调整数据删除策略：如果数据删除操作导致内存错误，可以考虑调整数据删除策略，例如使用更加高效的算法或者采用近似算法来进行数据删除。
总结起来，解决pyspark dataframe执行重复数据删除时遇到内存错误的问题，可以通过增加内存资源、分批处理数据、优化代码逻辑、增加数据处理节点、使用持久化存储、调整数据删除策略等方法来解决。具体的解决方案需要根据实际情况进行调整和优化。
相关·内容
我正在使用Pyspark2.0通过读取csv来创建一个DataFrame对象，使用：
data = spark.read.csv('data.csv', header=True)
我使用以下命令找到数据的类型
type(data)
结果是
pyspark.sql.dataframe.DataFrame
我正在尝试将数据中的一些列转换为LabeledPoint，以便应用分类。
from pyspark.sql.types import *
from pyspark.sql.functions import loc
from pyspark.mllib.regression
浏览 4提问于2016-09-08得票数 6
2回答
我正在使用EMR笔记本和pyspark和livy。
我从s3中读取数据，数据格式为拼图和字符串，并将其读入pyspark数据帧。大约有几个。200万行。当我执行连接操作时。我收到400会话处于非活动状态。为此，我已经将livy超时设置为5小时。
遇到错误:来自的无效状态代码'400‘，负载错误：“要求失败:会话未处于活动状态。”
浏览 1提问于2019-08-19得票数 1
我正在尝试迭代一个大型Pyspark Dataframe列中的所有不同值。当我尝试使用.distinct().collect()执行此操作时，即使只有两个不同的值，它也会发出“任务太大”警告。警告消息：
20/01/13 20:39:01 WARN TaskSetManager: Stage 0 contains a task of very large size (201 KB). The maximum recommended task size is 100 KB.
下面是一些示例代码：
from pyspark.sql import SparkSession
spark = Spar
浏览 1提问于2020-01-14得票数 1
我想使用Pyspark在数百个csv文件中读取，创建一个数据文件，即(粗略地)连接所有csv。因为每个csv都可以安装在内存中，但一次不超过一到两个，这似乎是一个很好的契合。我的策略不起作用，而且我认为这是因为我想在映射函数的内核函数中创建一个Pyspark dataframe，从而导致一个错误：
# initiate spark session and other variables
sc = SparkSession.builder.master("local").appName("Test").config(
"spark.driver.bindA
浏览 6提问于2022-10-03得票数 0
1回答
我正在使用内置Hadoop的PySpark 2.4.4处理一个大型数据集。数据集被分成多个压缩的CSV文件，我正在尝试使用PySpark的DataFrames报告各种列的聚合计数。在我要求PySpark使用以下命令将DataFrame写入磁盘之前，它一直处理得很好
df.write.csv("out.csv")
这会导致下面的警告淹没终端
WARN RowBasedKeyValueBatch: Calling spill() on RowBasedKeyValueBatch. Will not spill but return 0.
我尝试使用spark.executor.me
浏览 0提问于2020-02-23得票数 0
我正在尝试对pyspark数据帧执行一些操作。数据帧如下所示：
user domain1 domain2 ........ domain100 conversions
abcd 1 0 ........ 0 1
gcea 0 0 ........ 1 0
. . . ........ . .
. . .
浏览 0提问于2019-03-29得票数 4
3回答
我正在以火花DataFrame的形式读取CSV，并在其上执行机器学习操作。我一直得到一个Python序列化EOFError --知道为什么吗？我认为这可能是内存问题--即文件超出了可用内存--但大幅减少DataFrame的大小并没有防止EOF错误。
下面是玩具代码和错误。
#set spark context
conf = SparkConf().setMaster("local").setAppName("MyApp")
sc = SparkContext(conf = conf)
sqlContext = SQLContext(sc)
#read in 5
浏览 5提问于2016-04-12得票数 34
使用ApacheSpark2.0和pyspark，我有一个包含1000行数据的DataFrame，并希望将该DataFrame拆分/切片为2个单独的DataFrames；
第一个DataFrame应该包含前750个行
第二个DataFrame应该包含其余的250个行
注意:随机种子是不够的，因为我打算多次重复这种拆分方法，并且希望控制用于第一和第二DataFrame的数据。
我发现the (N)方法对于生成第一个结果很有用。
但我似乎无法找到获得第二个DataFrame的正确方法(或任何方法)。
任何指向正确方向的指示都将不胜感激。
提前谢谢。
Update：我现在设法通过重新
浏览 0提问于2016-08-15得票数 2
回答已采纳
我使用HiveContext (而不是SQLContext)在pyspark中创建了一个名为df的数据帧。
但是我发现在调用df.cache()之后，我将无法调用df.show()。例如：
>>> df.show(2)
+--------+-------------+--------+--------------+--------+-------+---------+--------+--------+-------------+--------+-----+
| bits| dst_ip|dst_port|flow_direction|in_iface|i
浏览 0提问于2016-05-16得票数 2
对于python dataframe，info()函数提供内存使用情况。在pyspark中有没有类似的东西？谢谢
浏览 3提问于2017-09-15得票数 11
1回答
我正在尝试将hql脚本转换为pyspark。我正在努力在groupby子句之后的聚合中实现case when语句的总和。例如：
dataframe1 = dataframe0.groupby(col0).agg(
SUM(f.when((col1 == 'ABC' | col2 == 'XYZ'), 1).otherwise(0)))
在pyspark中是可能的吗？我在执行这样的语句时遇到错误。谢谢
浏览 0提问于2016-11-23得票数 1
回答已采纳
2回答
我们有从多个源表读取数据并根据业务规则连接和应用映射的场景。在某些情况下，从几个表中读取的数据可以用于多个目标加载。那么，为了避免在运行不同的模块时多次读取相同的数据，有什么选择可以在不同的pyspark模块中使用相同的dataframe输出吗？
df1 = spark.sql(select * from table1)
df2 = spark.sql(select * from table2)
df_out = df1.join(df2, ['customer_id'], inner)
我想在pyspark_module1.py和pyspark_module2.py中使用d
浏览 0提问于2020-11-07得票数 0
1回答
用PySpark计算形状值
、、、、
我正在寻找一种方法来减少在我的大型数据集上计算SHAP值所需的计算时间(大约180米行，6个特性)，我遇到了这个，讨论了如何在SHAP上使用PySpark。
我是PySpark新手，我正在努力弄清楚如何使用本文中提供的代码片段来运行我的代码。
我现在使用下面的代码运行SHAP，其中还使用了X_values来适应我的隔离森林模型。
X_values = X.values
shap_values = explainer.shap_values(X_values)
这是文章的片段。
def calculate_shap(iterator: Iterator[pd.DataFrame]) ->
浏览 6提问于2022-07-28得票数 2
我想把PySpark dataframe (pyspark.sql.DataFrame)转换成Pandas数据格式。有一个内置的方法toPandas()，这是非常低效的(请阅读韦斯麦金尼关于这个问题的文章回到芬鲁里2017年和他在中的计算)。
与此同时，已经作出了一些努力，使这种转变更快。例如，Josh的函数。但是，这并没有帮助我，因为我希望从pysaprk.DataFrame传输+100万行到Pandas，这个解决方案对我不起作用。
幸运的是，正如2017年7月26日的所示，多亏了作者、和，由于Apache Arrow在Spark 2.3中的实现，toPandas()的功能得到了显著改善。尽
浏览 2提问于2017-09-07得票数 3
是否可以将pyspark数据帧传递给XGBClassifer，如下所示：
from xgboost import XGBClassifer
model1 = XGBClassifier()
model1.fit (df.select(features), df.select('label'))
如果不是，那么将pyspark dataframe应用到xgboost的最佳方式是什么？
非常感谢
浏览 0提问于2019-12-20得票数 0
1回答
我正在PySpark上进行一个项目，该项目需要处理大型数据集(大小约为2GB的多个.csv文件)。
假设我有两个数据A和B，分别对A和B分别执行一些转换。现在，让C是一些数据格式，这样：
C = A.join(B, A.key_1 == B.key_2, "full")
然后执行以下操作：
C.count()
PySpark是否在每次调用C.count()时都会重新执行对A和B*+* A+B的完全连接操作所执行的所有转换？还是在某个地方记住它们以加快动作速度？
假设有一个额外的数据数据D，它是C的转换， D.count()使PySpark重做A和B ?之间的连接
我注意到，当在
浏览 3提问于2022-11-14得票数 0
所以我在Kafka主题中有一些数据，我把这些流数据放到一个DataFrame中。我想要显示DataFrame中的数据：
import os
from kafka import KafkaProducer
from pyspark.sql import SparkSession, DataFrame
import time
from datetime import datetime, timedelta
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-1
浏览 3提问于2017-07-14得票数 5
回答已采纳
我试图使用pyspark.pandas API在数据库上构造数据的共生矩阵。
我试着用这个方法构造矩阵。
该代码在熊猫中运行良好，但在使用pyspark.pandas时出现了错误。
coocc = psdf.T.dot(psdf)
coocc
我得到了这个错误
TypeError: Unsupported type DataFrame
我查过医生了。
pyspark.pandas.DataFrame.dot()
以串联作为输入。
我尝试使用psdf.squeeze()将dataframe转换为系列化，但它并不将dataframe转换为串联，因为我的dataframe有多个列。
有没有办法将py
浏览 8提问于2022-10-14得票数 0
回答已采纳
2回答
我想绘制200 Gb的。我设法用datashader绘制/可视化了熊猫的数据。但是，我没有设法使用PySpark数据RAM (使用一个4节点集群，每个集群有8Gb内存)来完成它。不过，我可以做的是使用.toPandas()方法将PySpark数据into转换为Pandas数据into。但是这将在驱动节点上加载整个RAM中的数据(该节点没有足够的RAM来适应整个数据集)，因此不能利用Spark的分布式功能。
我也知道，只获取拾取和下降的经纬度和纬度将使数据到大约30 to。但这并不能改变问题。
我在GitHub这里创建了一个问题，
我认为Dask是另一种选择，但似乎还不支持转换PySpark d
浏览 3提问于2017-09-03得票数 2
回答已采纳
在PySpark和Synapse数据流中，我遇到了非常奇怪的错误。
我正在使用下面的查询读取JSON文件，但得到重复的列错误，即使没有重复的列。我可以使用其他工具和JSON验证器来读取它，也可以使用数据流，但不能在PySpark中读取。
PySpark查询如下：
df = (
spark.read.option("multiline", "true")
.options(encoding="UTF-8")
.load(
"abfss://<Container>]@<DIR>.
浏览 3提问于2021-11-25得票数 0
我想对排序后的Pyspark数据帧进行基于时间的训练测试拆分。假设前300行将在训练集中，下200行将在测试拆分中。 我可以用以下命令选择前300行： train = df.show(300) 但是如何从Pyspark dataframe中选择最后200行呢？
浏览 14提问于2019-03-13得票数 1
1回答
我在PySpark中创建了一个大块矩阵，名为mtm，具有85Kx85K维度。我想检查矩阵，以确保它是以我想要的方式创建的。我尝试过不同的路径，它们都失败了，退出代码143或92的内存问题都失败了。
我已经尝试过的选项: 1.将矩阵转换为rdd，并查看第一个条目：
mtm_coor = mtm.toCoordinateMatrix()
mtm_rdd = mtm_coor.entries
mtm_rdd.take(1)
将其保存到文本文件中
mtm_rdd.saveAsTextFile('./mtm.txt')
将其转换为DataFrame
mtm_df = mtm_rd
浏览 2提问于2017-08-22得票数 1
我有一个pyspark.sql.Row对象列表，如下所示：
[Row(artist=1255340), Row(artist=942), Row(artist=378), Row(artist=1180), Row(artist=813)]
从具有模式(id, name)的DataFrame中，我想输出id equals some artist in the given Row of list所在位置的行。什么才是正确的方法呢？
为了进一步澄清，我想这样做：select row from dataframe where row.id is in list_of_row_objects
浏览 1提问于2016-12-11得票数 1
1回答
在一个PySpark应用程序中，我试图将数据转换成熊猫，然后将结果写入csv文件。我就是这样做的：
df = df.toPandas().set_index("s").transpose()
df.coalesce(1).write.option("header", True).option("delimiter", ",").csv('dataframe')
在执行此脚本时，我会得到以下错误：
'DataFrame' object has no attribute 'coalesce
浏览 2提问于2017-09-28得票数 5
回答已采纳
我正在构建一个数据管道，它使用json格式的RESTApi数据并推送到Spark Dataframe。Spark版本: 2.4.4 但是得到的错误是 df = SQLContext.jsonRDD(rdd)
AttributeError: type object 'SQLContext' has no attribute 'jsonRDD' 代码： from pyspark import SparkConf,SparkContext
from pyspark.sql import SparkSession
from urllib import urlopen
浏览 27提问于2020-06-24得票数 2
回答已采纳
我正在尝试通过SparkNLP对文本数据进行主题建模和情感分析。我已经对数据集执行了所有预处理步骤，但在LDA中遇到错误。 Error 程序是： from pyspark.ml import Pipeline
from pyspark.ml.feature import StopWordsRemover, CountVectorizer, IDF
from pyspark.ml.clustering import LDA
from pyspark.sql.functions import col, lit, concat, regexp_replace
from pyspark.sql.ut
浏览 187提问于2021-04-22得票数 1
回答已采纳
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
spark = sqlContext.sparkSession
avg_calc = spark.read.csv("quiz2_algo.csv", header= True,inferSchema=True)
header = avg_calc.first()
no_header = avg_calc.subtract(header)
no_header
avg_calc包含2列，我正在尝试从这两列中删除第1行，但是我收到以下错误：
-------
浏览 1提问于2018-05-12得票数 0
我试图使用PySpark ()函数将toPandas数据trying转换为。然而，它的失败！
我的电火花数据表包含超过1500万条记录。这似乎是个问题吗？
from pyspark.sql import *
from pyspark.sql.functions import *
from pyspark.sql.types import *
import pandas as pd
df = sqlContext.table("schema.table")
df_pd=df.toPandas()
获取以下错误：遇到错误:来自 的无效状态代码'400‘，错误
浏览 0提问于2020-05-20得票数 1
我在HDFS上以Delta格式写入了数据。据我所知，Delta正在以拼接的形式存储数据，只是在它上面增加了一个具有高级功能的层。 但是，当使用Pyspark读取数据时，如果使用spark.read.parquet()或spark.read.format('delta').load()读取dataframe，我会得到不同的结果 df = spark.read.format('delta').load("my_data")
df.count()
> 184511389
df = spark.read.parquet("my_data
浏览 28提问于2021-01-27得票数 1
回答已采纳
我正在AWS EMR上运行一个5节点星火集群，每个大小为m3.xlarge (1个主站，4个从站)。我成功地运行了一个146 up bzip2压缩的CSV文件，最后得到了一个完美的聚合结果。
现在，我试图在这个集群上处理一个~5GB的bzip2 CSV文件，但是我收到了以下错误：
16/11/23 17:29:53警告TaskSetManager:在第6阶段(TID，xxx.compute.internal)中丢失的任务49.2 : ExecutorLostFailure (执行器16因运行任务之一而退出)原因:因超过记忆限制被纱线杀死的容器。使用10.4GB的10.4GB物理内存。考虑提高
浏览 8提问于2016-11-24得票数 64
2回答
我们都知道斯派克在内存中做计算。我只是对下面的内容感到好奇。
如果我从HDFS中在我的RDD外壳中创建了10个pySpark，这是否意味着所有这10个RDD的数据都将驻留在Spark内存中？
如果我不删除RDD，它会永远存在于记忆中吗？
如果我的数据集(文件)大小超过可用RAM大小，数据将存储在哪里？
浏览 5提问于2016-11-22得票数 21
回答已采纳
1回答
我正在尝试使用pyspark.ml库构建一个随机森林分类器，用于数据挖掘(，而不是RDD的mllib )。我是否必须使用文档中给出的管道？我只想建立一个简单的模型，
rf = RandomForestClassifier(labelCol = labs, featuresCol = rawdata)
我遇到以下错误
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/lib/spark/python/pyspark/__ini
浏览 0提问于2017-10-18得票数 0
1回答
是否有可能发送一个火花数据作为一个论据给熊猫UDF和得到一个熊猫的数据作为回报。下面是我正在使用的示例代码集，并且在调用函数时出错：
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.functions import pandas_udf
spark = SparkSession \
.builder \
.appName("PrimeBatch") \
.master("local[*]") \
.getOrCreate()
浏览 4提问于2020-11-26得票数 1
回答已采纳
局部模式运行
计算机内存：16‘s
计算机可用磁盘空间：500‘s
输入文件大小：50
错误消息：java.lang.OutOfMemoryError: Java堆空间
代码:在命令窗口中使用.\pyspark后，输入到windows命令提示符中的
码
from pyspark import SparkConf, SparkContext
import pyspark
rdd = sc.textFile('file:///myBigFile.xml')
rdd.persist(pyspark.StorageLevel.DISK_ONLY)
r
浏览 2提问于2019-01-31得票数 0
2回答
我有一个火花数据帧，有少量的字段。其中一些字段是巨大的二进制小块。整行的大小约为50 MB。
我正在将数据帧保存为拼花格式。我使用parquet.block.size参数控制行组的大小。
星火将生成一个拼花文件，但我将始终得到至少100行在一行组。这对我来说是个问题，因为块大小可能变成千兆字节，这与我的应用程序不太一样。
只要parquet.block.size的大小足以容纳100多行，就可以正常工作。
我将修改为MINIMUM_RECORD_COUNT_FOR_CHECK = 2，这解决了这个问题，但是，我发现没有任何配置值支持对这个硬编码常量进行调优。
是否有不同的/更好的方法来获得小于1
浏览 2提问于2018-01-09得票数 11
回答已采纳
我想使用Python将Dataframe (pyspark.pandas.Dataframe)作为一个Excel文件保存在Azure数据湖Gen2上。我已经切换到pyspark.pandas.Dataframe，因为它是从Spark3.2开始推荐的。
有一种名为to_excel (，文档)的方法允许将文件保存到ADL中的容器中，但我在文件系统访问协议方面遇到了问题。在同一个类中，我使用了使用abfss的to_csv和to_parquet方法，我想在excel中使用相同的方法。
因此，当我尝试这样保存它时，我使用：
import pyspark.pandas as ps
# Omit the d
浏览 32提问于2022-05-11得票数 0
回答已采纳
我知道PySpark DataFrames是不可变的，所以我想创建一个新列，该列是根据应用到PySpark DataFrame的现有列的转换生成的。我的数据太大了，无法使用collect()。
该列是唯一int列表(在给定列表中不重复int)，例如：
[1]
[1,2]
[1,2,3]
[2,3]
上面是一个玩具例子，因为我的实际DataFrame有一个最大长度为52个唯一ints的列表。我想要生成一个列，它遍历ints列表并为每个循环删除一个元素。要删除的元素将从所有列表中的唯一元素集合中删除，在本例中为[1,2,3]。
因此，对于第一次迭代：
删除元素1，结果如下：
[]
[2]
[2,3
浏览 9提问于2017-01-12得票数 1
回答已采纳
在运行pyspark作业时，随着输入的增加，我注意到我一直收到内存错误，如下所示.
ERROR cluster.YarnScheduler: Lost executor 12 on compute-2-10.local:
Container killed by YARN for exceeding memory limits. 1.5 GB of 1.5 GB
physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.
虽然它将成功地完成一个较小的输入，即使错误是通过执行重复的，但它最终会随着输入大小的
浏览 2提问于2017-10-07得票数 2
回答已采纳
我有一段python代码，它在本地运行在一个pandas数据帧中：
df_result = pd.DataFrame(df
.groupby('A')
.apply(lambda x: myFunction(zip(x.B, x.C), x.name))
我想在PySpark中运行它，但在处理pyspark.sql.group.GroupedData对象时遇到了问题。
我尝试过以下几种方法：
sparkDF
.groupby('A')
.agg(myFun
浏览 1提问于2016-10-13得票数 43
1回答
我列出我的数据帧以丢弃未使用的数据帧。首先，我使用下面的函数列出了我在其中一篇文章中找到的数据帧
from pyspark.sql import DataFrame
def list_dataframes():
return [k for (k, v) in globals().items() if isinstance(v, DataFrame)]
然后我试着从列表中删除未使用的。我在下面使用的代码
df2.unpersist()
当我再次列出时，df2仍然在那里。如何在pyspark上删除数据帧以获得一些内存？或者你还有其他的建议吗？谢谢。
浏览 0提问于2018-10-31得票数 4
回答已采纳
1回答
我正在使用readStream阅读kafka主题，并尝试在控制台上显示数据。
df = spark.readStream.format("kafka")\.option("kafka.bootstrap.servers", kafkaServer)\.option("subscribe", topic_name_read)\.option("includeHeaders", "true")\.option("startingOffsets", "latest")\.load()\.s
浏览 5提问于2022-03-30得票数 0
1回答
我有一个，它包含某些列值的副本，如下所示：
showing deptDF
+--------+----+------------+----------+--------+-----------+
|quantity|cost|participants|activity |category|id |
+--------+----+------------+----------+--------+-----------+
|4 |10 |2 |skiing |outdoor |8589934592 |
|4 |13 |3
浏览 1提问于2022-10-31得票数 0
注:这是写在齐柏林飞艇笔记本。
我一直试图复制源表的内容，将NULLS转换为0并将其插入目标表中。但是，我遇到了一个错误：
AnalysisException: u'\nUnsupported language features in query: INSERT INTO TABLE..
当在线阅读时，Spark似乎不支持INSERT INTO，但是，我一直在读到您可以使用dataframe并将其附加到表中。这是我的原始代码：
%pyspark
from pyspark.context import SparkContext
df = sqlContext.sql("SELE
浏览 0提问于2018-10-15得票数 1
回答已采纳
我需要在PySpark DataFrame中找到所有重复的记录。以下是示例数据集：
# Prepare Data
data = [("A", "A", 1), \
("A", "A", 2), \
("A", "A", 3), \
("A", "B", 4), \
("A", "B", 5), \
("A", "C", 6), \
("A",
浏览 6提问于2022-11-30得票数 0
回答已采纳
1回答
我用的是Pyspark 我的输入数据如下所示。 COL1|COL2
|TYCO|130003|
|EMC |120989|
|VOLVO|102329|
|BMW|130157|
|FORD|503004|
|TYCO|130003| 我已经创建了DataFrame并查询重复项，如下所示。 from pyspark.sql import Row
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appName("Test") \
.getOrCreat
浏览 8提问于2019-09-24得票数 0
回答已采纳
1回答
我正在为星火数据集构建管道，并收到以下错误消息：
'DataFrame' object has no attribute 'transform'
因为我的dataframe类是：
<class 'pyspark.sql.dataframe.DataFrame'>
而不是
<class 'pyspark.sql.DataFrame'>
如何获得正确的类，以便在管道中使用.transform()函数？以下是文档中的内容：
浏览 1提问于2022-06-11得票数 -1
回答已采纳
我有一个正在运行的Google Dataproc集群，并且正在向它提交一个PySpark作业，该作业从Google Cloud Storage (945MB CSV文件，400万行-->总共需要48秒才能读入)读取一个文件到PySpark dataframe，并对该数据帧应用一个函数(parsed_dataframe = raw_dataframe.rdd.map(parse_user_agents).toDF() -->大约需要4到5秒)。
然后，我必须将这些修改后的结果作为GZIP的CSV或Parquet文件保存回Google Cloud Storage。我还可以将这些修改后的
浏览 8提问于2017-08-01得票数 4
我目前正在自学Spark programming，并试图用PySpark重新编写一个现有的Python应用程序。然而，我仍然对如何在PySpark中使用常规Python对象感到困惑。
我了解Spark中的分布式数据结构，如RDD、DataFrame、Datasets、vector等。Spark有自己的转换操作和动作操作，如.map()、.reduceByKey()来操作这些对象。但是，如果我在PySpark中创建传统的Python数据对象，比如数组、列表、元组或字典，会怎么样呢？它们将只存储在我的驱动程序节点的内存中，对吗？如果我把它们转换成RDD，我还能用典型的Python函数做操作吗？
如
浏览 34提问于2017-03-01得票数 1
回答已采纳
1回答
我对pyspark/bigdata有点陌生，所以这可能不是一个好主意，但我有大约一百万个单独的CSV文件，每个文件都与一些元数据相关联。我想要一个pyspark dataframe所有元数据字段的列，但也有一个列，其条目是(整个) CSV文件与每组元数据相关联。
我现在不在工作，但我几乎记得确切的代码。我试过一个玩具的例子，比如
outer_pandas_df = pd.DataFrame.from_dict({"A":[1,2,3],"B":[4,5,6]})
## A B
## 0 1 4
## 1 2 5
## 2 3 6
如果你这样
浏览 2提问于2016-11-17得票数 0
我试图使用Pyspark连接插入到现有的Mysql表中，但是我得到了以下错误
File "<stdin>", line 1, in <module>
File "/usr/hdp/current/spark2-client/python/pyspark/sql/context.py", line 384, in sql
return self.sparkSession.sql(sqlQuery)
File "/usr/hdp/current/spark2-client/python/pyspark/sql/se
浏览 0提问于2020-07-17得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
