如何在pyspark数据帧中使用write.partitionBy时删除重复项？
在pyspark中，使用write.partitionBy时删除重复项可以通过以下步骤实现：
- 首先，导入必要的模块和函数：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col- 创建SparkSession对象：
spark = SparkSession.builder.appName("PartitionByExample").getOrCreate()- 读取数据源文件并创建数据帧：
df = spark.read.csv("path/to/input/file.csv", header=True, inferSchema=True)请将"path/to/input/file.csv"替换为实际的数据源文件路径。
- 使用write.partitionBy方法将数据帧按照指定的列进行分区，并保存到指定的输出路径：
df.write.partitionBy("column1", "column2").format("parquet").mode("overwrite").save("path/to/output")请将"column1"和"column2"替换为实际的列名，将"path/to/output"替换为实际的输出路径。
- 删除重复项：
df = spark.read.parquet("path/to/output")
df = df.dropDuplicates()请将"path/to/output"替换为上一步保存数据的输出路径。
完整的代码示例如下：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("PartitionByExample").getOrCreate()
df = spark.read.csv("path/to/input/file.csv", header=True, inferSchema=True)
df.write.partitionBy("column1", "column2").format("parquet").mode("overwrite").save("path/to/output")
df = spark.read.parquet("path/to/output")
df = df.dropDuplicates()这样，你就可以在pyspark数据帧中使用write.partitionBy时删除重复项了。
关于腾讯云相关产品和产品介绍链接地址，由于要求不能提及具体品牌商，建议你参考腾讯云官方文档或咨询腾讯云的技术支持团队，获取与pyspark数据帧操作相关的产品和服务信息。
相关·内容
我有一个数据帧，如下所示： |------------|-----------|---------------|---------------||------------|-----------|---------------|---------------| 我想根据名称和类型对此数据帧进行分区并将其保存到磁盘，如下所述 在文件夹中 /output/Roger/A
浏览 15提问于2020-10-08得票数 0
回答已采纳
当我用groupBy()替换spark数据帧上的distinct()时，我在我的pyspark代码中看到了很大的性能改进。但是我不能理解背后的原因。整个意图是从数据帧中删除行级重复项。我尝试在谷歌上搜索groupBy()和distinct()在pyspark中的实现，但没有找到。
有没有人能给我解释一下或者给我指出正确的方向？
浏览 0提问于2018-09-11得票数 6
2回答
我有大量的最终结果数据，这些数据在我感兴趣的列中分布得并不均匀。当我通过分区直接写入时，每个分区具有与spark.sql.shuffle.partitions相同数量的文件。这会导致拥挤分区中的每个文件都非常大(以GB为单位)，但在其他一些分区中，文件大小实际上很小(甚至以KB为单位)。有没有办法改变每个分区的文件数？-------------------+----------+df.write.partitionBy("category").parquet(output_dir)
浏览 2提问于2018-12-05得票数 0
1回答
我列出我的数据帧以丢弃未使用的数据帧。首先，我使用下面的函数列出了我在其中一篇文章中找到的数据帧然后我试着从列表中删除未使
浏览 0提问于2018-10-31得票数 4
回答已采纳
2回答
我有一个pyspark数据帧：----------------------------------------------------------------------------------------
在matricule数组中，如果我删除AZ字符串，我会有重复
浏览 28提问于2020-02-26得票数 2
回答已采纳
2回答
我正在尝试删除数据帧中的所有重复项。removeDuplicates = data.drop_duplicates()
第一行应该保留，所以第二次出现的县1也就是重复的应该被删除。但是，当我对此运行drop duplicates时，没有任何反应
编辑:不要紧，它打印的是不同的数据
浏览 0提问于2021-02-17得票数 0
1回答
我有一个PySpark Dataframe，它包含一个ArrayType(StringType())列。此列包含需要删除的数组中的重复字符串。假设我的数据文件名为df，我的列名为arraycol。arraycol_without_dupes", F.remove_dupes_from_array("arraycol"))
我的直觉是，这有一个简单的解决方案，但是在浏览堆栈溢出15分钟之后，我没有发现比爆炸列、删除完整数据帧
浏览 1提问于2019-01-14得票数 2
回答已采纳
我有一个从kafka主题中读取的spark流数据帧，我想在每次解析新记录时删除过去5分钟的重复数据。我知道dropDuplicates(["uid"])函数，但我不确定如何在特定的历史时间间隔内检查重复项。我的理解是： df = df.dropDuplicates(["uid"]) 要么处理当前(微)批处理读取的数据，要么处理当前正在进入内存的“任何”数据。有
浏览 31提问于2020-04-21得票数 0
回答已采纳
在Spark中，使用pyspark，我有一个包含重复项的数据帧。我想用像电子邮件和mobile_phone这样的多重规则来删除重复数据。这是我在python 3中的代码： from pyspark.sql import Row
df = sc.parallelizer
浏览 10提问于2018-12-27得票数 2
回答已采纳
如何在join操作后将具有相同列名的数据帧写入csv文件。目前，我正在使用以下代码。write.format('com.databricks.spark.csv').save('/home/user/output/',header = 'true')将在“/home/user/output”中写入数据帧"dfFinal“.But它在
浏览 0提问于2018-10-03得票数 6
我有两个数据帧，我正在合并它。在合并时，它应该删除重复项。但是对于帧1中的一个重复行，它应该只删除帧2中的一个重复行，即使有两个这样的行，如下面的df1： colA colB colC 1 1 2 1 2 3 1 1 2 结果： c
浏览 32提问于2021-01-23得票数 1
回答已采纳
我有以下表格作为RDD：1 y1 y1 n2 n我想从Value中删除所有的重复项。输出应如下所示：1 y2 y在pyspark中工作时，输出应该是如下所示的键值对列表：
[(u'1',u'n'),(u'2',u'n'
浏览 4提问于2014-09-18得票数 14
回答已采纳
我是新来Pyspark的。我有一个Pyspark dataframe，我想根据id和时间戳列删除重复项。然后，我想将重复id的读取值替换为null。我不想用熊猫。如下所示： 数据帧： id reading timestamp12 16700 2018-03-22 09:00:00.000
2
浏览 13提问于2020-01-08得票数 2
回答已采纳
1回答
我有两个数据帧，如下root |-- name: string(nullable = true)nullable = true) |-- school: string(nullable = true)
由于这些数据帧包含不同学校的学生的详细信息，因此为了标识唯一的行，应该始终使用id和学校列。我想根
浏览 0提问于2021-10-04得票数 0
3回答
这通常是在一个大的步骤之后，或者缓存一个我想要多次使用的状态之后。似乎当我第二次对我的数据帧调用cache时，一个新的副本被缓存到内存中。在我的应用程序中，这会导致在扩展时出现内存问题。尽管在我当前的测试中，给定的数据帧最大大约为100MB，但中间结果的累积大小超出了executor上分配的内存。请参见下面的小示例，该示例演示了此行为。C2'])df.show()
spark_co
浏览 9提问于2016-04-28得票数 39
回答已采纳
1回答
我有一个数据帧，如下所示我想删除其中包含"id"和"ac"的列，并保留其他列我应该如何在这里的列名上使用regexep？
浏览 1提问于2020-02-06得票数 0
如下所示df.registerTempTable('temp_table')
我想按行对我的数据帧df进行切片(例如，相当于熊猫数据帧中的df.iloc[0:4000], df.iloc[4000:8000]等)，因为我想将每个小块转换为熊猫数据帧，以便稍后处理。我只知道如何使用sample
浏览 5提问于2016-09-08得票数 4
我有一个pyspark脚本，它将被airflow重复执行。df.write.mode('overwrite').parquet(s3_url)
它会完全清除目录(删除目录)。我是不是误解了“覆盖”模式？当我用上面的代码尝试这样做时，它删除了现有的目录。
浏览 7提问于2021-05-12得票数 1
1回答
我想知道如何在Spark (Pyspark)中实现以下内容+--+---++--+---++--+---++--+---++--+---++--+---++--+---+-------++--+---+-------++--+---+-------+
|3 |7.0|
浏览 50提问于2015-12-16得票数 38
回答已采纳
我正在尝试使用子集(drop_duplicates=‘’，keep=False)在我的数据帧中删除重复项。显然，它在我的Jupyter Notebook中工作正常，但当我试图通过终端以.py文件的形式执行时，我得到了以下错误： Traceback (most recent call last):
File"/home/source/fork/PySpark_Analytics&#x
浏览 110提问于2019-06-20得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
