Pyspark:复制连接2个不同行数的数据帧
Pyspark是一个基于Python的Spark编程接口，它提供了用于大规模数据处理和分析的高级API。Pyspark可以与Spark集群一起使用，利用分布式计算的优势来处理大规模数据。
复制连接两个不同行数的数据帧是指将两个具有不同行数的数据帧进行连接操作。在Pyspark中，可以使用join函数来实现数据帧的连接操作。具体步骤如下：
- 导入必要的模块和函数：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col- 创建SparkSession对象：
spark = SparkSession.builder.getOrCreate()- 创建两个数据帧：
df1 = spark.createDataFrame([(1, 'A'), (2, 'B'), (3, 'C')], ['id', 'value1'])
df2 = spark.createDataFrame([(1, 'X'), (2, 'Y'), (3, 'Z'), (4, 'W')], ['id', 'value2'])- 使用join函数进行连接操作：
joined_df = df1.join(df2, on='id', how='inner')在上述代码中，使用join函数将df1和df2按照'id'列进行内连接操作，生成一个新的数据帧joined_df。
连接操作的参数说明：
- on：指定连接的列名，这里使用'id'列进行连接。
- how：指定连接的方式，这里使用内连接（inner join）。
连接操作的结果是一个新的数据帧joined_df，它包含了两个数据帧的连接结果。
Pyspark相关产品和产品介绍链接地址：
- Apache Spark：Spark是一个快速、通用的大数据处理引擎，Pyspark是其Python编程接口。
- Tencent Cloud EMR：腾讯云的弹性MapReduce（EMR）是一种大数据处理和分析服务，支持Pyspark等多种编程语言和工具。
- Tencent Cloud CVM：腾讯云的云服务器（CVM）提供了高性能、可扩展的计算资源，可用于部署Spark集群和运行Pyspark程序。
请注意，以上提到的腾讯云产品仅作为示例，其他云计算品牌商也提供类似的产品和服务。
相关·内容
在PySpark中是否有与pandas info()方法等效的方法？
我正在尝试获取有关PySpark中数据帧的基本统计信息，例如:列数和行数、空值数、数据帧大小
pandas中的Info()方法提供了所有这些统计信息。
浏览 4提问于2017-06-08得票数 6
我有两个相同行数的pyspark dataframe，但它们没有任何公共列。因此，我使用monotonically_increasing_id()将新列添加到这两个列中
from pyspark.sql.functions import monotonically_increasing_id as mi
id=mi()
df1 = df1.withColumn("match_id", id)
cont_data = cont_data.withColumn("match_id", id)
cont_data = cont_data.join(df1,df1.ma
浏览 2提问于2017-06-03得票数 1
我有一个运行在EMR上的pyspark应用程序，我想监控它的一些指标。例如，计数已加载、已保存的行数。目前我使用count操作来提取值，这显然会减慢应用程序的运行速度。我在想，是否有更好的选择来从数据帧中提取这些指标？我使用的是pyspark 2.4.5
浏览 0提问于2021-01-26得票数 0
我有一个pyspark数据帧，我想把它分成多个相等记录的数据帧。我在AWS EMR上执行此任务，不支持pandas或numpy。
浏览 22提问于2019-12-22得票数 0
1回答
我有一个SparkSQL数据帧和2Dnumpy矩阵。它们具有相同的行数。我打算将numpy矩阵中的每个不同数组作为新列添加到现有的PySpark数据帧中。这样，添加到每一行的列表是不同的。 例如，PySpark数据帧如下所示 | Id | Name |
| ------ | ------ |
| 1 | Bob |
| 2 | Alice |
| 3 | Mike | numpy矩阵是这样的 [[2, 3, 5]
[5, 2, 6]
[1, 4, 7]] 产生的预期数据帧应该如下所示 | Id | Name | custo
浏览 16提问于2019-10-05得票数 0
1回答
我有两个包含GB级数据的大型pyspark数据帧df1和df2。第一个数据帧中的列是id1、col1。第二个数据帧中的列是id2、col2。数据帧具有相等的行数。此外，id1和id2的所有值都是唯一的。此外，id1的所有值都恰好对应于一个值id2。 为。前几个条目是用于df1和df2区域的 df1： id1 | col1
12 | john
23 | chris
35 | david df2： id2 | col2
23 | lewis
35 | boon
12 | cena 所以我需要连接key id1和id2上的两个数据帧。df = df1.
浏览 9提问于2020-06-10得票数 2
回答已采纳
1回答
我有一个pyspark管道(包含估算和一个机器学习模型)和一个pandas数据帧。我是否可以在不将其转换为Pyspark dataframe的情况下将管道应用于此pandas数据帧？如果不可能，我如何有效地使用pyspark管道来生成对pandas数据帧的预测？
浏览 3提问于2021-09-13得票数 1
这是用于使用Spark2.3.2的Python/PySpark。我正在寻找最佳实践方法，将一个数据框架的列复制到另一个数据框架，使用PySpark对一个非常大的10+十亿行数据集(按年/月/日平均划分)。每一行都有120列要转换/复制。输出数据帧将被写入另一组文件中，日期分区。
示例模式是：input DFinput (colA，colB，colC)和输出DFoutput (X，Y，Z)
我想将DFInput复制到DFOutput，如下所示(colA => Z，colB => X，colC => Y)。
在Python 2.3+中这样做的最佳实践是什么？是否应该对每个列使用D
浏览 1提问于2018-12-19得票数 5
在pyspark (和一般的python )中执行两个数据帧的滚动连接时，我遇到了很多问题。我希望将两个pyspark数据帧通过它们的ID和最近日期反向连接在一起(这意味着第二个数据帧中的日期不能晚于第一个数据帧中的日期)
Table_1：
Table_2：
期望的结果：
从本质上讲，我知道SQL查询可以做spark.sql(“查询”)等任何事情。我尝试了几种在spark环境下不起作用的方法。谢谢!
浏览 1提问于2020-08-08得票数 2
我们使用的是与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
回答已采纳
我想在两个数据帧之间执行左连接，但是列并不完全匹配。第一个数据帧中的联接列相对于第二个数据帧有一个额外的后缀。
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
回答已采纳
1回答
我有一个pyspark数据帧作为
DOCTOR | PATIENT
JOHN | SAM
JOHN | PETER
JOHN | ROBIN
BEN | ROSE
BEN | GRAY
并且需要按行连接患者姓名，这样我就可以得到如下输出：
DOCTOR | PATIENT
JOHN | SAM, PETER, ROBIN
BEN | ROSE, GRAY
有没有人能帮我在pyspark中创建这个数据帧？
提前谢谢。
浏览 13提问于2017-01-22得票数 18
回答已采纳
1回答
我有一个csv数据文件，可以加载到pyspark中：
spark = SparkSession.builder.master("local").appName("MYAPP").getOrCreate()```
df = spark.read.csv( path = csvfilepath, sep="|", schema=my_schema, nullValue="NULL", mode="DROPMALFORMED")```
检查数据帧的行数得到大约2000万行。
df.count()
我在拼图中重新存储了我
浏览 7提问于2020-05-06得票数 0
1回答
我有一个数据帧列表，在列表的每个位置上，我有一个数据帧，我需要将所有这些数据帧合并到一个数据帧中。在我使用之前，这是在PySpark中完成的 dataframe_new =pd.concat(listName) 解决方案1 from pyspark.sql.types import *
import pyspark.sql
from pyspark.sql import SparkSession, Row
customSchema = StructType([
StructField("col1", StringType(), True),
Str
浏览 12提问于2021-01-28得票数 0
2回答
我有一个大约有十亿行的PySpark数据帧。我想要取每2000个值的平均值，比如指数为0-1999的行数的平均值，指数为2000-3999的行数的平均值，等等。我该怎么做呢？或者，我也可以为每2000个平均值取10个值，比如指数为0-9的行数的平均值，指数为2000-2009的行数的平均值，等等。这样做的目的是对数据进行下采样。我目前没有索引行，所以如果我需要它，我该怎么做呢？
浏览 12提问于2018-03-15得票数 1
1回答
我有两个几乎相同的Pyspark数据帧:相同的行数和row_id，相同的模式，但每行的某些列上的值不同。 我想确定每一行的列是什么。 示例： 数据帧A id fname lname email
1 Michael Jackson mj@yahoo.com
2 Roger Moore rm@rocketmail.com
3 Angela Merkel am@dw.de 数据帧B id fname lname email
1 Michael Jordan mj@yahoo.com
2 Gordon Moore rm@rocketmail
浏览 22提问于2020-03-23得票数 1
回答已采纳
1回答
我正在做一个使用spark的项目。在某些阶段，我需要在单个数据帧中合并或连接3个数据帧。这些数据帧来自spark sql表，我使用了联合函数，它已经合并了两个表中具有相同编号的列，但我也需要合并不相等的列值。我现在很困惑，有没有办法在pyspark中合并或连接不相等的基于列的数据帧，请指导我
浏览 13提问于2016-09-22得票数 0
我在Databricks上创建了一个PySpark DataFrame。
%python
# File location and type
file_location = "/FileStore/tables/file.csv"
file_type = "csv"
# CSV options
infer_schema = "true"
first_row_is_header = "true"
delimiter = ";"
# The applied options are for CSV files. Fo
浏览 1提问于2018-10-05得票数 0
我有一个pyspark数据帧 spark = SparkSession\
.builder\
.appName("NPS_TF")\
.getOrCreate()
df2 = spark.createDataFrame([
("unknown", 1,2,3)
], ["Assign", "xs[0]","xs[1]","xs[2]"])
df2.limit(1).show() 如何将部分dataframe列名和第一行
浏览 10提问于2019-02-13得票数 0
回答已采纳
我是新来齐柏林飞艇的。我有一个用例，其中我有一个熊猫数据帧。我需要使用齐柏林飞艇的内置图表来可视化集合。我在这里没有明确的方法。我的理解是，如果数据是RDD格式，我们可以使用zeppelin可视化数据。首先，我尝试将pandas的数据帧转换为spark的数据帧，但失败了
%pyspark
import pandas as pd
from pyspark.sql import SQLContext
print sc
df = pd.DataFrame([("foo", 1), ("bar", 2)], columns=("k", "v
浏览 0提问于2015-10-06得票数 17
我想用两个不同的条件和一个dataframe : df = dataframecol1，col2在Pyspark中创建to列。
col1 =基于df_A::MTAV = df_B::CODE将df_B与df_A：：MTAV连接起来
col2 =基于df_A::MTAP = df_B::CODE将df_B与df_A：：MTAP连接起来
如何在数据帧中创建两列？
浏览 1提问于2021-12-20得票数 -1
回答已采纳
我有一个PySpark RDDs的字典，正在尝试将它们转换为数据帧，将它们保存为变量，然后连接它们。当我尝试将其中一个RDDs转换为数据帧时，出现以下错误：
File "./spark-1.3.1/python/pyspark/sql/types.py",
line 986, in _verify_type
"length of fields (%d)" % (len(obj), len(dataType.fields)))
ValueError: Length of object (52) does not match with length of
浏览 1提问于2015-07-08得票数 0
我已经使用以下代码创建了数据框： import pyspark
from pyspark.sql import functions as F
sc = pyspark.SparkContext()
spark = pyspark.sql.SparkSession(sc)
data = [('A', 'B', 1), ('A', 'B', 2), ('A', 'C', 1)]
columns = ['Column1', 'Col
浏览 12提问于2020-06-15得票数 0
2回答
我想使用PySpark提取SharePoint列表数据。我不确定Sharepoint列表数据和存储。我想以PySpark数据帧的形式读取SharePoint列表数据。
我尝试过Python Libraies: Sharepy Slum和许多其他库
浏览 2提问于2019-05-24得票数 2
如何在sql like分区上按条件连接两个DataFrames？我实际上需要连接两个数据帧，以便对于每个组(基于一个列变量)，我与其他表进行外部连接。
例如，我有以下两个数据帧：
df1：
df2：
所需的输出类似于：
我曾尝试使用Pyspark的Window操作符，但无法使用，因为它不能用于窗口上的连接。
任何帮助都将不胜感激。
浏览 12提问于2019-07-25得票数 0
我正在使用PySpark连接、过滤并将大型数据写入csv。
在每个过滤器连接或写入之后，我用df.count()来计算行数。
然而，计算行数意味着重新加载数据并重新执行各种操作。
如何计算每个不同操作期间的行数，而不像使用df.count()那样重新加载和计算？
我知道缓存函数可能是一个不重新加载和重新计算的解决方案，但是我正在寻找另一个解决方案，因为它并不总是最好的解决方案。
提前谢谢你！
浏览 5提问于2022-04-21得票数 0
在链接中，用户可以在Spark3.2中的PySpark之上与熊猫合作。是否需要很长时间才能将PySpark数据帧转换为PySpark熊猫数据框架？
我知道将PySpark数据帧转换为熊猫数据框架需要很长时间。
浏览 9提问于2022-03-02得票数 1
回答已采纳
3回答
我有一个很大的pyspark数据帧，我正在对它执行一些转换，并与其他数据帧连接。我想调查转换和连接是否成功，以及数据帧是否看起来像是预期的，但我如何显示数据帧的一小部分。
我试过很多东西。
df.show(5)
和
df.limit(5).show()
但是，我尝试的所有操作都需要大量的作业，从而导致性能低下。我可以旋转一个非常大的集群，但是有没有一种方法可以快速地只获取数据帧的一小部分？
浏览 0提问于2019-11-20得票数 2
我有两个数据帧，我需要使用具有两个连接谓词的非相等连接(即不等式连接)将它们连接在一起。
一个数据帧是直方图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不返回输出数据。
from pyspark.sql import *
import pyspark.sql.functions as F
from pyspark.sql.functions import col
spark = SparkSession.builder.master("local").appName("test").getOrCreate()
file_path="C:\\bigdata\\pipesep_data\\Sales_ny.csv"
df=spark.read.form
浏览 5提问于2022-06-06得票数 0
回答已采纳
在连接两个数据帧时，我在pyspark中遇到了问题。第一个dataframe是一个单列的dataframe "zipcd"，第二个是四个列的dataframe。 每当我试图连接两个dataframe时，问题就会出现，因为Pyspark在我的新dataframe中返回我，关于zipcd的一列，它的所有值都是相同的(第一行在所有行中都是重复的，它不是这样的)。 例如： Zip.select("Zip").show()
+------------+
| Zip|
+------------+
| 6.0651002E8|
| 6.0623002E8|
浏览 30提问于2021-07-01得票数 1
我有一个包含2列(CPID和PluginDuration)的spark数据帧。我需要找到数据帧中每个CPID的最大pluginDuration和平均pluginDuration。 为CPID AN04773数据帧返回的行数在行数下面返回： df.filter('CPID = "AN04773"').show(10)
Result:
+-------+--------------+
| CPID|PluginDuration|
+-------+--------------+
|AN04773| 1.933333333|
|AN04773| 13
浏览 47提问于2021-11-01得票数 1
回答已采纳
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回答
我正在尝试将以下SQL查询转换为pyspark： SELECT COUNT( CASE WHEN COALESCE(data.pred,0) != 0 AND COALESCE(data.val,0) != 0
AND (ABS(COALESCE(data.pred,0) - COALESCE(data.val,0)) / COALESCE(data.val,0)) > 0.1
THEN data.pred END) / COUNT(*) AS Result 我现在在PySpark中的代码是： Result = data.select(
count(
浏览 13提问于2020-12-28得票数 0
回答已采纳
我有一个pyspark数据帧df，它保存了大量的rows.Once列是lat-long。我想从经纬度中找到州的名称。我使用下面的代码 import reverse_geocoder as rg
new_df = df_new2.toPandas()
list_long_lat = a["lat_long"].tolist()
result = rg.search(list_long_lat)
state_name=[]
for each_entry in result:
state_name.append(each_entry["admin2"])
浏览 12提问于2020-07-01得票数 0
回答已采纳
1回答
对于每个key_id和date_month，输入pyspark数据帧都有一行。对于一个随机的key_id，它看起来像这样 +--------+-------------+---------+---------+
| key_id | date_month | value_1 | value_2 |
+--------+-------------+---------+---------+
| 1 | 2019-02-01 | 1.135 | 'a' |
| 1 | 2019-03-01 | 0.165 | 'b'
浏览 46提问于2019-04-30得票数 1
1回答
从Server 2008开始的从条款文件中简要地提到了3个连接提示及其基本机制：
Reduce
复制
再分配
然而，似乎没有多少关于何时有必要使用它们的信息。
看来，它们可以与哈希、循环和合并一起使用，这些都是为了这个问题的目的已经被理解的。
文件中的有关章节：
对于SQL数据仓库和并行数据仓库，这些连接提示适用于两个发行版不兼容列上的内部联接。它们可以通过限制查询处理过程中发生的数据量来提高查询性能。SQL数据仓库和并行数据仓库允许的连接提示如下：
减少了连接右侧表要移动的行数，以便使两个发行版不兼容的表兼容。减少提示也称为半连接提示。
复制使连接列中的值从联接左侧的表中复制到所有节点。右边
浏览 0提问于2019-01-24得票数 8
回答已采纳
2回答
在PySpark中有什么特定的方法可以像我们在r中那样绑定两个数据帧吗？
示例：
数据帧1有10列。
数据帧2有1列
我需要在PySpark中同时绑定数据帧和作为一个数据帧。
浏览 1提问于2017-08-30得票数 3
2回答
我正在使用EMR笔记本和pyspark和livy。
我从s3中读取数据，数据格式为拼图和字符串，并将其读入pyspark数据帧。大约有几个。200万行。当我执行连接操作时。我收到400会话处于非活动状态。为此，我已经将livy超时设置为5小时。
遇到错误:来自的无效状态代码'400‘，负载错误：“要求失败:会话未处于活动状态。”
浏览 1提问于2019-08-19得票数 1
有没有可能在不使用Hive或Phoenix或Hortonworks提供的spark-Hbase连接器的情况下直接以Pyspark Dataframes的形式读取Hbase表？
我对Hbase比较陌生，找不到一个直接的Python示例来将Hbase表转换为Pyspark数据帧。我看到的大多数示例都是用Scala或Java编写的。
浏览 2提问于2017-10-05得票数 1
有没有人可以在不转换xlsx或xls文件的情况下让我知道，我们如何才能将它们作为spark数据帧读取？
我已经尝试使用pandas读取数据，然后尝试转换为spark dataframe，但得到错误，错误是
错误：
Cannot merge type <class 'pyspark.sql.types.DoubleType'> and <class 'pyspark.sql.types.StringType'>
代码：
import pandas
import os
df = pandas.read_excel('/dbfs/Fil
浏览 25提问于2019-06-03得票数 8
1回答
我在pyspark中有一个dataframe，df有一个数组字符串类型的列，所以我需要生成一个包含列表头部的新列，还需要包含尾列表连接的其他列。 这是我的原始数据帧： pyspark> df.show()
+---+------------+
| id| lst_col|
+---+------------+
| 1|[a, b, c, d]|
+---+------------+
pyspark> df.printSchema()
root
|-- id: integer (nullable = false)
|-- lst_col: array (nulla
浏览 29提问于2020-01-24得票数 0
回答已采纳
2回答
我想要加载存储在S3 AWS文件夹结构中的所有拼图文件。
文件夹结构如下：<code>D0</code>
我想要的是一次读取所有的拼图文件，所以我希望PySpark读取2019年所有可用的月份和日期的所有数据，然后将其存储在一个数据帧中(这样您就可以获得一个包含2019年所有日期的连接/联合数据帧)。
我被告知这些是分区的文件(虽然我不确定这一点)。
这在PySpark中是可能的吗?如果是的话，是如何实现的？
当我尝试<code>D1</code>时，它起作用了。但是，当我想使用<code>D2</code>查看Spark
浏览 70提问于2019-12-04得票数 0
1回答
我有两个数据帧，它们是从两个csv文件中读取的。
+---+----------+-----------------+
| ID| NUMBER | RECHARGE_AMOUNT|
+---+----------+-----------------+
| 1|9090909092| 30|
| 2|9090909093| 30|
| 3|9090909090| 30|
| 4|9090909094| 30|
+---+----------+--------------
浏览 1提问于2018-10-03得票数 1
当我用groupBy()替换spark数据帧上的distinct()时，我在我的pyspark代码中看到了很大的性能改进。但是我不能理解背后的原因。整个意图是从数据帧中删除行级重复项。
我尝试在谷歌上搜索groupBy()和distinct()在pyspark中的实现，但没有找到。
有没有人能给我解释一下或者给我指出正确的方向？
浏览 0提问于2018-09-11得票数 6
我之前发布了这个问题，并得到了一些使用PySpark的建议。
下面的压缩文件()包含一个名为data的文件夹，其中包含大约130,000个csv文件。我想把它们合并到一个单独的数据帧中。我有16 of的RAM，当我访问前几百个文件时，我一直在耗尽RAM。这些文件的总大小只有300-400MB的数据。
如果您打开任何csv文件，您可以看到它们都具有相同的格式，第一列用于日期，第二列用于数据系列。
所以现在我改用PySpark，但是我不知道连接所有文件的最有效的方法是什么，对于pandas数据帧，我会像这样连接单个帧的列表，因为我想让它们在日期上合并：
bigframe = pd.concat(
浏览 14提问于2020-02-17得票数 2
回答已采纳
我在pyspark 2.4.0中有一个dataframe df，我想使用sampleBy获取它的样本。它包含一个列category，我有一个这样的dict作为示例： dict = {
"cat_A" : 0.1,
"cat_B" : 0.5,
"cat_C" : 0.4
} 我希望我的样本有5000行，如果我像这样使用函数sampleBy：df_sampled = df.sampleBy(col = 'category', fractions = fraction, seed = 1)，得到的数据帧具有每个类别的正确比例，但似
浏览 198提问于2021-01-08得票数 1
我有一个pyspark数据帧，并且我想在其自身上执行笛卡尔连接。我在pyspark中使用了下面的函数 # Cross Join
m_f_1 = m_f_0.withColumnRenamed('value', 'value_x').crossJoin(m_f_0.withColumnRenamed('value', 'value_y')).drop(m_f_0.area).drop(m_f_0.id) 我面临的问题是，还有一个列名类型，我希望在执行交叉连接时将其重命名。该怎么做呢？ m_f_0数据帧是： id v
浏览 24提问于2019-05-20得票数 1
回答已采纳
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
回答已采纳
我的pyspark数据帧中有多语言字符。在将数据写入bigquery之后，它向我显示了奇怪的字符，因为它的默认编码方案(utf-8)。如何使用pyspark / dataproc将Bigquery中的编码更改为ISO_8859_1？
浏览 0提问于2020-05-06得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
云直播活动推荐
腾讯云开发者
