使用pyspark查找每个相应列的两个数据帧上的值的差异
可以通过以下步骤实现：
- 导入必要的库和模块：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col- 创建SparkSession对象：
spark = SparkSession.builder.appName("DataFrameDiff").getOrCreate()- 创建两个数据帧DataFrame1和DataFrame2：
data1 = [("John", 25, "USA"), ("Alice", 30, "Canada"), ("Bob", 35, "UK")]
data2 = [("John", 25, "USA"), ("Alice", 28, "Canada"), ("Bob", 35, "UK")]
df1 = spark.createDataFrame(data1, ["Name", "Age", "Country"])
df2 = spark.createDataFrame(data2, ["Name", "Age", "Country"])- 使用join操作将两个数据帧按照相应列进行连接：
joined_df = df1.join(df2, on=["Name", "Age"], how="inner")- 使用select操作选择需要比较的列，并计算差异：
diff_df = joined_df.select("Name", "Age", (col("Country_x") != col("Country_y")).alias("CountryDiff"))- 显示差异结果：
diff_df.show()以上代码将显示每个相应列的差异结果，其中"CountryDiff"列将显示True或False，表示两个数据帧上对应列的值是否相同。
对于pyspark的使用，可以参考腾讯云的Apache Spark产品介绍：Apache Spark产品介绍
注意：以上答案仅供参考，具体实现方式可能因环境和需求而异。
相关·内容
1回答
有没有办法只读取两个版本的deltaLake之间的增量？尤其是在Synapse中
我试过了
%%pyspark
import delta
from pyspark.sql.functions import col, asc
df_delta = spark.read.format("delta") \
.option("readChangeFeed", "true") \
.option("startingVersion", 0) \
.option("endingVersion", 1) \
.
浏览 0提问于2021-05-24得票数 1
我需要帮助来连接两个数据帧，其中一个是空的，另一个有数据。你能告诉我如何在pyspark中做到这一点吗？ 我使用的是熊猫： 假设df2为空，而df1有一些记录。 df2 = pd.concat([df2, df1]) 但是如何在pyspark中执行这个操作呢？ df1:
+--------------------+----------+---------+
| Programname|Projectnum| Drug|
+--------------------+----------+---------+
|Non-Oncology Phar...|SR0480-0
浏览 6提问于2020-12-02得票数 0
我有两个数据帧，我需要使用具有两个连接谓词的非相等连接(即不等式连接)将它们连接在一起。
一个数据帧是直方图DataFrame[bin: bigint, lower_bound: double, upper_bound: double]
另一个数据帧是观察值DataFrame[id: bigint, observation: double]的集合
我需要确定每个观察值落入直方图的哪个柱状图中，如下所示：
observations_df.join(histogram_df,
(
(observations_df.observation >= histogram_d
浏览 1提问于2018-10-02得票数 1
2回答
在PySpark中有什么特定的方法可以像我们在r中那样绑定两个数据帧吗？
示例：
数据帧1有10列。
数据帧2有1列
我需要在PySpark中同时绑定数据帧和作为一个数据帧。
浏览 1提问于2017-08-30得票数 3
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
回答已采纳
我有两个数据帧df1和df2，我只想在结果中不匹配的列。我试着使用SQL来做，但是SQL返回所有列，而不是一列。
df1
col1|col2|col3
a b c
1 2 3
df2
col1|col2|col3
a b e
1 2 3
我想要的是如果它能返回
df3
col3
是否可以在pyspark中做do，或者我必须通过从两个数据帧中选择每一列然后进行比较来进行比较？
浏览 5提问于2018-08-20得票数 0
1回答
我有以下表格的火花数据格式：
from pyspark.sql.functions import *
from pyspark.sql.types import *
schema_sdf_consistent = StructType([
StructField("A", DoubleType(), True),
StructField("B", DoubleType(), True),
StructField("C", DoubleType(), True),
])
sdf_consistent
浏览 2提问于2022-05-26得票数 0
我有一个问题，当我们开始录制mac屏幕。我只需要采取唯一的帧，而不是重复的帧。可以使用Macos accelerate框架进行过滤吗？
浏览 4提问于2020-10-15得票数 0
1回答
我有一个火花源数据帧(Df1)，它由10K行组成，数据帧看起来像- id mobile_no value
1 1111111111 .43
2 2222222222 .54
3 3333333333 .03
4 4444444444 .22 另一个火花源数据帧(df2)由100k记录组成，看起来像- mobile_no gender
912222222222 M
914444444444 M
91
浏览 25提问于2019-06-21得票数 1
回答已采纳
1回答
我有两个数据帧，它们是从两个csv文件中读取的。
+---+----------+-----------------+
| ID| NUMBER | RECHARGE_AMOUNT|
+---+----------+-----------------+
| 1|9090909092| 30|
| 2|9090909093| 30|
| 3|9090909090| 30|
| 4|9090909094| 30|
+---+----------+--------------
浏览 1提问于2018-10-03得票数 1
2回答
我有两个pyspark数据帧，我想检查第一个数据帧列值是否存在于第二列dataframe.If第一个数据帧列值不存在于第二个数据帧列中，我需要确定这些值并将其写入list.Is有没有更好的方法来使用pyspark来处理这种情况？感谢您的回复。 df[Name].show()
Java
Oracle
.NET
df1[Name].show()
Oracle
Scala
.NET
python
浏览 37提问于2020-09-03得票数 0
回答已采纳
2回答
我需要找出pyspark dataframe中所有列中0的百分比。如何在数据帧中找到每一列的零计数？
附言:我尝试过将数据帧转换为pandas数据帧，并使用了value_counts。但是，对于大型数据集来说，推断它的观察值是不可能的。
浏览 0提问于2018-08-20得票数 4
有一系列数值变量的pyspark数据帧。 例如 我的dataframe有一个从1到100的列值。 1-10 - group1<== 1到10的列值应包含group1作为值11-20 - group2。。。91-100 group10 如何使用pyspark dataframe实现这一点？
浏览 15提问于2019-04-10得票数 1
回答已采纳
我有两个pyspark数据帧tdf和fdf，其中fdf比tdf大得多。这些数据帧的大小每天都在变化，我不知道它们的大小。我想从fdf中随机挑选数据来组成一个新的数据帧rdf，其中rdf的大小大约等于tdf的大小。目前我有以下几行代码： tdf = tdf.count()
fdf = fdf.count()
sampling_fraction = float(tdf) / float(fdf)
rdf = fdf(sampling_fraction, SEED) 这些行产生正确的结果。但是当fdf的大小增加时，fdf.count()需要几天的时间才能完成。你能推荐另一种在PySpark中更快的
浏览 15提问于2021-01-30得票数 0
回答已采纳
有没有办法检查当前行中的所有列值是否小于pandas数据帧(整个数据帧)中所有前一行(直到当前行)中的相应列值，并相应地创建一个值为1或0的新列？
浏览 0提问于2017-06-15得票数 1
1回答
我正在根据列的值填充数据帧。数据帧中的行可以重复。
df = DataFrame({'A' : [5,6,3,4], 'B' : [1,2,3, 5]})
out[16]:
A B
0 5 1
1 6 2
2 3 3
3 4 5
df[df['A'].isin([3, 6])]
Out[17]:
A B
1 6 2
2 3 3
这是意料之中的。我需要的是不同的东西。我需要有重复的行。这怎么可能呢？
df[df['A'].isin([3, 6,3])]
浏览 0提问于2016-05-11得票数 1
我正在研究PySpark。我有一个数据框，我需要将其转储为JSON文件，但JSON文件应具有以下格式，例如- {"Column 1": [9202, 9202, 9202, ....], "Column 2": ["FEMALE", "No matching concept", "MALE", ....]} 因此，每列应该有一个键，相应的值应该包含该列中的所有值的列表 我尝试将其转换为Pandas数据帧，然后在将其作为JSON转储之前转换为字典，并成功地完成了转换，但由于数据量非常大，我想直接在PySpark数据
浏览 14提问于2019-12-18得票数 1
回答已采纳
我有另一种解决方案，但我更喜欢使用PySpark 2.3来实现它。
我有一个二维PySpark数据框架，如下所示：
Date | ID
---------- | ----
08/31/2018 | 10
09/31/2018 | 10
09/01/2018 | null
09/01/2018 | null
09/01/2018 | 12
我希望通过查找过去最接近的值来替换ID空值，或者如果该值是空的，则向前看(如果它再次为空，则设置一个默认值)。
我想象过用.withColumn添加一个新列，并使用一个UDF函数来查询数据帧本身。
类似于伪代码的内容(并不完美，但它是主要思想)：
浏览 0提问于2019-02-02得票数 0
回答已采纳
1回答
我正在做一个使用spark的项目。在某些阶段，我需要在单个数据帧中合并或连接3个数据帧。这些数据帧来自spark sql表，我使用了联合函数，它已经合并了两个表中具有相同编号的列，但我也需要合并不相等的列值。我现在很困惑，有没有办法在pyspark中合并或连接不相等的基于列的数据帧，请指导我
浏览 13提问于2016-09-22得票数 0
2回答
我有一个硬币抛出的矢量x <- c("H","H","T","H","T","T","H","H","H","H","T")。我想从上面的向量中构造一个包含两个变量(模式，长度)的数据帧。
Pattern字段将类似于{H H T，H T，T，H T}，相应的长度将为{2,1,0,4}
我的意图是计算连续的正面，直到抛硬币时出现一条尾巴。
浏览 1提问于2016-05-10得票数 4
1回答
我有一个非常庞大的数据集，我需要使用pyspark dataframe。请参考数据的简化版本： product_type series_no product_amount date
514 111 20 2020/01/01 (YYYY/MM/DD)
514 111 30 2020/01/02
514 111 40 2020/01/03
514
浏览 30提问于2021-01-23得票数 0
回答已采纳
1回答
我有一个PySpark DataFrame，df1，它看起来像：
CustomerID CustomerValue CustomerValue2
12 .17 .08
我有第二个PySpark DataFrame，df2
CustomerID CustomerValue CustomerValue
15 .17 .14
16 .40 .43
18 .86 .09
我想要取两个数据帧的余弦相似度。并拥有这样的东西
Customer
浏览 2提问于2018-09-28得票数 1
我正在尝试连接两个这样的pyspark数据帧 joined = df.join(df1,on=["date"],how='left').select([col('df.'+xx) for xx in df.columns] + [col('df1.daily_net_payment_sum'),col('df1.daily_net_payment_avg')]) 但它会导致 An error was encountered:
"cannot resolve '`df.cust_no`' g
浏览 15提问于2021-04-17得票数 0
回答已采纳
1回答
我有两个spark数据帧df1和df2。我尝试将它们写到两个不同的文件路径。有人能告诉我，写操作是同步进行还是异步进行？这是因为它们是两个不同的数据帧写入两个不同的路径，写入是同时发生，还是必须等到它完成df1写出后才开始写df2？
示例代码：
更新添加的导入库：
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueConte
浏览 15提问于2020-02-17得票数 0
我们使用的是与Spark 1.3.1接口的PySpark库。
我们有两个数据帧，documents_df := {document_id, document_text}和keywords_df := {keyword}。我们希望连接这两个数据帧，并使用{document_id, keyword}出现在document_df.document_text字符串中的条件，返回具有keyword_df.keyword对的结果数据帧。
例如，在PostgreSQL中，我们可以使用以下形式的ON子句来实现这一点：
document_df.document_text ilike '%' ||
浏览 2提问于2015-10-16得票数 8
回答已采纳
我有220 GB的数据。我已经将其作为两列读取到spark dataframe中: JournalID和Text。现在，我的数据帧中缺少27行。
使用NGram类，我在dataframe中添加了另外两个列Unigram和Bigram，其中包含文本列中的单字和双字。然后，我使用一元和二元语法列上的pyspark的TF和IDF类计算TFIDF，并将其作为另一列添加到dataframe中。
现在我为dataframe中的每一行都有了journalID和TFIDF vector。我想将支持向量机应用于所有类型的核，以TFIDF向量作为特征，以JournalID作为标签。由于多类支持向量机不存在于pys
浏览 1提问于2018-12-17得票数 0
1回答
在MatLab中，我被赋予了创建一个程序的任务：
模拟了两个六边模N次轧制的实验，并对两个轧制的数值进行了求和，并绘制了得到这些数值的频率图。
则可以打印滚动结果中最频繁和最不频繁的结果之间的百分比差异。
我已经想出了如何做第一部分：
numberOfDice = 2; %Number of dice to be rolled
numberOfDots = 6; %Number of max. dots allowed on a die face
numberOfRolls = 100000; %Number of times the die(s) are rolled
%#
浏览 2提问于2016-09-19得票数 0
回答已采纳
1回答
我打算在下面的difflib.SequenceMatcher数据帧上使用PySpark ()。
tech.show()
+-----------------------------+----------------------+
| concat_tech |Vendor_product |
+-----------------------------+----------------------+
|AWS Cloud Administration |AWS Cloud Map |
|Grad Portal
浏览 1提问于2021-03-05得票数 1
回答已采纳
在pyspark (和一般的python )中执行两个数据帧的滚动连接时，我遇到了很多问题。我希望将两个pyspark数据帧通过它们的ID和最近日期反向连接在一起(这意味着第二个数据帧中的日期不能晚于第一个数据帧中的日期)
Table_1：
Table_2：
期望的结果：
从本质上讲，我知道SQL查询可以做spark.sql(“查询”)等任何事情。我尝试了几种在spark环境下不起作用的方法。谢谢!
浏览 1提问于2020-08-08得票数 2
我目前在本地的windows10系统上使用pyspark。pyspark代码运行得相当快，但将pyspark数据帧保存为csv格式需要花费大量时间。
我正在将pyspark数据帧转换为pandas，然后将其保存到csv文件中。我还尝试使用write方法来保存csv文件。
Full_data.toPandas().to_csv("Level 1 - {} Hourly Avg Data.csv".format(yr), index=False)
Full_data.repartition(1).write.format('com.databricks.spark.
浏览 126提问于2019-08-01得票数 5
在对数据集运行pyspark中的ALS算法后，我遇到了一个最终的数据帧，如下所示 ? 推荐的列是数组类型，现在我想拆分这一列，我的最终数据帧应该如下所示 ? 谁能建议我，哪个pyspark函数可以用来形成这个数据帧？ 数据帧的模式 root
|-- person: string (nullable = false)
|-- recommendation: array (nullable = true)
| |-- element: struct (containsNull = true)
| | |-- ID: string (nullable =
浏览 61提问于2021-07-11得票数 0
回答已采纳
我在pyspark dataframe中有一个count列，如下所示：
id Count Percent
a 3 50
b 3 50
我想要一个结果数据帧为：
id Count Percent CCount CPercent
a 3 50 3 50
b 3 50 6 100
我不能使用熊猫数据帧，因为数据库太大了。我找到了指向窗口分区的答案，但我没有这样的列作为分区依据。请大家用pyspark dataframe告诉我怎么做。注意: pysp
浏览 5提问于2017-03-20得票数 0
1回答
大熊猫在火花源中计数
、、、
目前正在尝试将我从pandas创建的脚本转换为pyspark，我有一个数据帧，其中包含以下形式的数据： index | letter
------|-------
0 | a
1 | a
2 | b
3 | c
4 | a
5 | a
6 | b 我想创建以下数据帧，其中存储了字母的每个实例的出现次数，例如，第一次看到"a“时，它的出现次数为0，第二次为1，第三次为2： index | letter | occurrence
------|--------|-----------
0 | a | 0
1
浏览 4提问于2019-06-20得票数 0
回答已采纳
我希望跨多个列应用插入符号的confusionMatrix函数，将结果保存在数据帧中。
我有两个相同格式的数据框架--“实际值”和“预测”。它们看起来是这样的：
x1 x2 x3 x4
N N S E
E W E E
N W E E
等等。对于许多行，每列有4个级别，并有相当多的列。
我想使用confusionMatrix比较这两个数据帧的对应列。我可以逐列执行此操作，例如，从第1列开始：
confusionMatrix(predictions[,1], actuals[,1])
但是，我不想逐一查看每一列，而是将conf
浏览 0提问于2018-09-25得票数 1
回答已采纳
我正在寻找一种有效的方法，将一些映射函数应用于数据帧中的每一对元素。例如： records = spark.createDataFrame(
[(1, 'a'), (2, 'b'), (3, 'c'), (4, 'd')], \
['id', 'val'])
records.show()
+---+---+
| id|val|
+---+---+
| 1| a|
| 2| b|
| 3| c|
| 4| d|
+---+---+ 我想获取a，b，c，d值，并将它们
浏览 6提问于2020-11-20得票数 0
我想在两个数据帧之间执行左连接，但是列并不完全匹配。第一个数据帧中的联接列相对于第二个数据帧有一个额外的后缀。
from pyspark import SparkContext
import pyspark.sql.functions as f
sc = SparkContext()
df1 = sc.parallelize([
['AB-101-1', 'el1', 1.5],
['ABC-1020-1', 'el2', 1.3],
['AC-1030-1', 'el3'
浏览 1提问于2017-08-08得票数 3
回答已采纳
假设我在Pyspark中有一个名为table_df的数据框
sid | date | label
------------------
1033| 20170521 | 0
1033| 20170520 | 0
1033| 20170519 | 1
1033| 20170516 | 0
1033| 20170515 | 0
1033| 20170511 | 1
1033| 20170511 | 0
1033| 20170509 | 0
.....................
数据帧table_df在不同的行中包
浏览 6提问于2018-02-23得票数 0
在PySpark中是否有与pandas info()方法等效的方法？
我正在尝试获取有关PySpark中数据帧的基本统计信息，例如:列数和行数、空值数、数据帧大小
pandas中的Info()方法提供了所有这些统计信息。
浏览 4提问于2017-06-08得票数 6
如何在循环中创建一个pyspark DataFrame？在这个循环中，在每次迭代中，我输出2个值print(a1,a2)。现在，我想将所有这些值存储在一个pyspark数据帧中。
浏览 26提问于2021-01-12得票数 0
2回答
我有一个像这样的PySpark数据文件：
+------+------+
| A| B|
+------+------+
| 1| 2|
| 1| 3|
| 2| 3|
| 2| 5|
+------+------+
我想对表进行查找，看看是否存在特定的行。例如，对于A = 2，B = 5的测试，代码应该返回True，对于A = 2，B = 10，代码应该返回False。
我试过这个：
df[(df['A'] == 1) & (df['B'] == 2)].rdd.isEmp
浏览 2提问于2021-02-10得票数 3
回答已采纳
我有一个在多个列中有Nulls的PySpark数据帧。我正在尝试用相应列的最大数字+ 1填充空值。这个想法是，在为每个数字分类列创建“缺失”类别时，在尝试模拟SAS中的一个热编码过程时，返回并在一次热编码后将这些值标记为“无”。
例如，male中的“null”应为“3”。因此，公列的最大数目+1= 3。然后是4，其他列的最大数目是6。
示例数据：
+----------------+------+-----+-----+
| MS| male| rate|level|
+----------------+------+-----+-----+-
|
浏览 0提问于2019-08-22得票数 1
我有csv数据，并使用read_csv创建了Pandas dataframe，并强制所有列为字符串。然后，当我尝试从Pandas数据帧创建Spark数据帧时，我得到了下面的错误消息。
from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql.types import *
z=pd.read_csv("mydata.csv", dtype=str)
z.info()
<class 'pandas.core.frame.DataFrame'>
浏览 6提问于2016-08-06得票数 33
回答已采纳
1回答
我一直试图在我的库中查看一个视频，并使用python和OpenCV检测冻结的帧。
有一个给出了答案，但我不太明白。答案如下：
冻结帧:计算两个连续帧中每个像素的HSV/RGB绝对差，并确定用于检测冻结帧的最大允许差值。
在视频中检测超过3秒的冻结帧的最佳方法是什么？
非常感谢。
浏览 3提问于2020-07-04得票数 1
回答已采纳
2回答
我有一个使用Pandas数据框架用Python编写的朴素贝叶斯分类器，现在我需要它在PySpark中使用。我在这里的问题是，我需要每个专栏的特性重要性。在查看PySpark ML文档时，我找不到它的任何信息。
有没有人知道我是否能通过朴素的贝叶斯星火MLlib获得特性的重要性？
使用Python的代码如下。使用.coef_检索功能的重要性。
df = df.fillna(0).toPandas()
X_df = df.drop(['NOT_OPEN', 'unique_id'], axis = 1)
X = X_df.values
Y = df['NO
浏览 0提问于2019-01-09得票数 1
回答已采纳
1回答
我必须用两个任务在Airflow中运行两个模块。每个任务都有一个执行一些spark操作的PySpark模块。第二个模块使用在前一个会话中创建的数据帧并继续其操作。同样的SparkSession初始化，如何实现同样的效果？我尝试使用getActiveSession()，但由于任务1作业已完成，因此无法工作，因此当任务2运行时，会创建一个新的spark会话。 - [root@ ..dags]# cat tmp_spark_1.py
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("
浏览 32提问于2021-11-23得票数 0
在连接两个数据帧时，我得到了重复的数据帧，其中一个键是小数，另一个是字符串。Spark似乎正在将小数转换为字符串，这将导致科学记数表达式，但随后将原始结果以十进制形式显示出来。我找到了一种通过直接转换为字符串的解决方法，但这似乎很危险，因为在没有警告的情况下会创建重复项。这是一个bug吗？我如何才能检测到这种情况何时发生？ 下面是Spark 2.4上的pyspark演示： >>> from pyspark.sql.functions import *
>>> from pyspark.sql.types import *
>>> df1
浏览 15提问于2019-06-08得票数 3
我之前发布了这个问题，并得到了一些使用PySpark的建议。
下面的压缩文件()包含一个名为data的文件夹，其中包含大约130,000个csv文件。我想把它们合并到一个单独的数据帧中。我有16 of的RAM，当我访问前几百个文件时，我一直在耗尽RAM。这些文件的总大小只有300-400MB的数据。
如果您打开任何csv文件，您可以看到它们都具有相同的格式，第一列用于日期，第二列用于数据系列。
所以现在我改用PySpark，但是我不知道连接所有文件的最有效的方法是什么，对于pandas数据帧，我会像这样连接单个帧的列表，因为我想让它们在日期上合并：
bigframe = pd.concat(
浏览 14提问于2020-02-17得票数 2
回答已采纳
我有一个有几列的pyspark dataframe col1 col2 col3
---------------------
1. 2.1. 3.2
3.2. 4.2. 5.1 我想将三个函数f1(x), f2(x), f3(x)分别应用于数据帧的对应列，以便获得 col1 col2 col3
-------------------------------
f1(1.) f2(2.1.) f3(3.2)
f1(3.2.) f2(4.2.) f3(5.1) 我尽量避免为每一列定义一个udf，所以我的
浏览 49提问于2020-07-29得票数 0
回答已采纳
1回答
我有一个PySpark数据帧，它有一个复杂的列，请参考下列值：
ID value
1 [{"label":"animal","value":"cat"},{"label":null,"value":"George"}]
我想在PySpark dataframe中添加一个新列，它基本上将它转换为一个字符串列表。如果Label为null，则字符串应包含value；如果label不为null，则string应为label:value。因此，对于上面的示例数据帧，输出应如下所示：
ID
浏览 0提问于2021-02-09得票数 0
这里有两个数据帧：
预期的数据帧：
+------+---------+--------+----------+-------+--------+
|emp_id| emp_city|emp_name| emp_phone|emp_sal|emp_site|
+------+---------+--------+----------+-------+--------+
| 3| Chennai| rahman|9848022330| 45000|SanRamon|
| 1|Hyderabad| ram|9848022338| 50000| SF|
|
浏览 0提问于2017-06-03得票数 21
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
