通过计算现有列的比率在Pyspark DataFrame中创建新列
在Pyspark DataFrame中，可以通过计算现有列的比率来创建新列。具体步骤如下：
- 首先，导入必要的库和模块：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col- 创建一个SparkSession对象：
spark = SparkSession.builder.getOrCreate()- 加载数据并创建DataFrame：
data = [(1, 10, 20), (2, 15, 25), (3, 20, 30)]
df = spark.createDataFrame(data, ["id", "numerator", "denominator"])这里的data是一个包含三个元组的列表，每个元组表示一行数据，其中第一个元素是id，第二个元素是分子，第三个元素是分母。["id", "numerator", "denominator"]是DataFrame的列名。
- 使用withColumn()函数计算比率并创建新列：
df = df.withColumn("ratio", col("numerator") / col("denominator"))这里使用col()函数来引用DataFrame中的列，通过除法运算符计算比率，并将结果存储在名为"ratio"的新列中。
- 打印DataFrame的内容：
df.show()输出结果如下：
+---+---------+-----------+-----+
|id |numerator|denominator|ratio|
+---+---------+-----------+-----+
|1 |10 |20 |0.5 |
|2 |15 |25 |0.6 |
|3 |20 |30 |0.67 |
+---+---------+-----------+-----+这样就成功在Pyspark DataFrame中创建了一个新列，该新列存储了现有列的比率。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云计算服务：https://cloud.tencent.com/product/cvm
- 腾讯云数据库服务：https://cloud.tencent.com/product/cdb
- 腾讯云人工智能服务：https://cloud.tencent.com/product/ai
- 腾讯云物联网服务：https://cloud.tencent.com/product/iotexplorer
- 腾讯云移动开发服务：https://cloud.tencent.com/product/mobdev
- 腾讯云存储服务：https://cloud.tencent.com/product/cos
- 腾讯云区块链服务：https://cloud.tencent.com/product/baas
- 腾讯云元宇宙服务：https://cloud.tencent.com/product/ue
- 腾讯云安全服务：https://cloud.tencent.com/product/safe
相关·内容
1回答
动态地在多个列上操作
、、、
在pyspark中，假设我有名为'a1','a2','a3'...'a99'的列的dataframe，如何对每个列应用操作来动态创建具有新名称的新列？
例如，获取新列(如sum('a1') as 'total_a1' , ... sum('a99') as 'total_a99' )。
浏览 0提问于2019-02-28得票数 1
1回答
我将df作为一个pyspark.sql.dataframe.DataFrame导入到Databricks中。在这个df中，我有3列(我已经证实它们是字符串)，我希望将它们连接起来。我试过先使用一个简单的"+“函数。
df["fullname"] = df["firstname"] + df["middlename"] + df["lastname"]
但是我一直收到错误"'DataFrame‘对象不支持项分配“。因此，我试图在每一列后面添加.astype(str)，但没有结果。最后，我尝试简单地添加另一列，
浏览 1提问于2022-12-02得票数 0
我在一个PySpark DataFrame中有两列，我想在填充空值(而不是就地)后取这两列的比率。目前我的DataFrame如下所示： +----+----+---+----+----+----+----+
|Acct| M1D|M1C| M2D| M2C| M3D| M3C|
+----+----+---+----+----+----+----+
| B| 10|200|null|null| 20|null|
| C|1000|100| 10|null|null|null|
| A| 100|200| 200| 200| 300| 10|
+----+----+---+
浏览 17提问于2019-12-19得票数 1
回答已采纳
我已经在现有的数据框中添加了一个新列，但它没有反映在数据帧中。
customerDf.withColumn("fullname",expr("concat(firstname,'|',lastname)"))
customerDf.show() # it's showing existing old df records without new columns.
如果我们可以将数据帧分配给另一个数据帧，我们就可以看到结果
test = customerDf.withColumn("fullname",expr("
浏览 4提问于2021-01-26得票数 0
10回答
我有一个火花DataFrame (使用PySpark 1.5.1)，并希望添加一个新的列。
我尝试过以下几种方法，但都没有成功：
type(randomed_hours) # => list
# Create in Python and transform to RDD
new_col = pd.DataFrame(randomed_hours, columns=['new_col'])
spark_new_col = sqlContext.createDataFrame(new_col)
my_df_spark.withColumn("hours"
浏览 13提问于2015-11-12得票数 179
回答已采纳
1回答
如何将Vectors.dense列添加到pyspark中？
import pandas as pd
from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.ml.linalg import DenseVector
py_df = pd.DataFrame.from_dict({"time": [59., 115., 156., 421.], "event": [1, 1, 1, 0]})
sc = SparkContext(master="loc
浏览 1提问于2018-04-14得票数 1
回答已采纳
给定以下DataFrame：
import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local").appName("test").getOrCreate()
df = spark.createDataFrame([['a',1],['b', 2],['a', 3]], ['category', 'value'])
df.
浏览 2提问于2018-09-10得票数 7
回答已采纳
1回答
我有一台pyspark DataFrame
a = [
('Bob', 562),
('Bob',880),
('Bob',380),
('Sue',85),
('Sue',963)
]
df = spark.createDataFrame(a, ["Person", "Amount"])
我需要创建一个列来散列Amount并返回金额。问题是我不能使用UDF，所以我使用了一个映射函数。
df.rdd.map(lambda x: hash(x[
浏览 1提问于2018-04-17得票数 11
回答已采纳
Pyspark API提供了除median之外的许多聚合函数。Spark 2附带了approxQuantile，它给出了近似的分位数，但精确的中位数计算起来非常昂贵。对于Spark Dataframe中的一列值，是否有更多的Pyspark方法来计算中值？
浏览 3提问于2016-08-03得票数 3
回答已采纳
我的df运行Spark2.1.0有一个问题，它有几个字符串列创建为一个来自提供这个.summary()的Hive的SQL查询：
DataFrame[summary: string, visitorid: string, eventtype: string, ..., target: string]。
如果我只运行df.groupBy("eventtype").count()，它就能工作，我就可以得到DataFrame[eventtype: string, count: bigint]
在使用show df.groupBy('eventtype').count()
浏览 6提问于2017-07-19得票数 1
回答已采纳
我正在将一些用熊猫编写的代码转换为PySpark。该代码有许多for循环，用于根据用户指定的输入创建可变数量的列。
我使用的是Spark 1.6.x，其中包含以下示例代码：
from pyspark.sql import SQLContext
from pyspark.sql import functions as F
import pandas as pd
import numpy as np
# create a Pandas DataFrame, then convert to Spark DataFrame
test = sqlContext.createDataFrame(pd.D
浏览 1提问于2016-10-15得票数 9
回答已采纳
1回答
将函数应用于两列
、、
假设我有两个PySpark DataFrames df1和df2。
df1= 'a'
1
2
5
df2= 'b'
3
6
我希望为每个df2['b']找到最接近的df1['a']值，并将最近的值作为df1中的一个新列添加。
换句话说，对于df1['a']中的每个值df1['a']，我希望找到一个实现min(abx(x-y)) for all y in df2['b']
浏览 2提问于2016-11-02得票数 10
回答已采纳
我有一个spark dataframe，它有两列: name，age，如下：
[Row(name=u'Alice', age=2), Row(name=u'Bob', age=5)]
数据帧是使用以下命令创建的
sqlContext.createDataFrame()
接下来我需要做的是从外部的'csv‘文件中添加第三列'UserId’。外部文件有几列，但我只需要包括第一列，即'UserId'：
两个数据源中的记录数相同。我在windows操作系统上使用的是独立的pyspark版本。最终结果应该是一个包含三列的新数据帧:
浏览 9提问于2016-09-16得票数 0
回答已采纳
我希望按一列分组，然后找到另一列的最大值。最后，显示基于此条件的所有列。然而，当我使用我的代码时，它只显示2列，而不是所有列。
# Normal way of creating dataframe in pyspark
sdataframe_temp = spark.createDataFrame([
(2,2,'0-2'),
(2,23,'22-24')],
['a', 'b', 'c']
)
sdataframe_temp2 = spark.createDataFrame([
浏览 0提问于2020-01-19得票数 0
回答已采纳
1回答
在下面的代码中，我试图从管道the创建一个DataFrame：
print type(simulation)
sqlContext.createDataFrame(simulation)
print语句打印如下：
<class 'pyspark.rdd.PipelinedRDD'>
但是，在下一行中，我得到了以下错误：
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 3.0 failed 1 times, most recent failur
浏览 2提问于2016-07-06得票数 1
回答已采纳
我正在尝试使用PySpark将一个分数应用于星火PySpark。让我们假设我在Spark之外构建了一个简单的回归模型，并希望将模型中创建的系数值映射到DataFrame中的各个列，以创建一个新列，即每个不同源列的和乘以各个系数。我知道Spark中有许多用于建模的实用程序，但我想了解如何实现这种“蛮力”方法。我还知道DataFrames/RDD是不可变的，因此必须创建一个新的DataFrame。
以下是一些供参考的伪代码：
#load example data
df = sqlContext.createDataFrame(data)
df.show(5)
dfmappd.select("
浏览 4提问于2016-03-01得票数 1
回答已采纳
我希望找到每列的1%le和99%le，并删除各自的pyspark dataframe行。谢谢
浏览 2提问于2017-05-18得票数 0
1回答
如何在不立即触发计算的情况下，高效地将列除以其在星火DataFrame中的自身和？
假设我们有一些数据：
import pyspark
from pyspark.sql import SparkSession, Window
import pyspark.sql.functions as spf
spark = SparkSession.builder.master('local').getOrCreate()
data = spark.range(0, 100)
data # --> DataFrame[id: bigint]
我想在这个数据框架上创建一个名为“规
浏览 0提问于2018-01-31得票数 2
回答已采纳
我想要创建一个函数，该函数从PySpark中的左联接创建一个新列，详细说明一个列中的值是否匹配或不匹配另一个dataframe逐行的列。
例如，我们有一个PySpark dataframe (d1)具有列ID和名称，另一个PySpark dataframe (d2)具有相同的列- ID和Name。
我试图创建一个连接这两个表的函数，并创建一个新列，如果两个数据文件中存在相同的ID，则创建一个显示“True”或“False”的新列。
到目前为止，我有这个
def doValuesMatch(df1, df2):
left_join = df1.join(df2, on='ID&#
浏览 3提问于2021-12-11得票数 0
1回答
我有两个数据帧：data和rules。
>>>data >>>rules
vendor rule
0 googel 0 google
1 google 1 dell
2 googly 2 macbook
在计算了每个供应商和规则之间的Levenshtein相似度之后，我尝试将两个新列添
浏览 0提问于2017-08-25得票数 8
回答已采纳
我目前有一个pyspark dataframe，其中一列包含我想要使用我编写的返回信息字符串的函数查找的数字行。我知道最简单的方法是使用withCoulmn并定义一个UDF来从旧的列创建一个新列，但是我的函数不能将它注册为UDF.Is的方式使我可以基于旧列的值创建一个新的dataframe而不需要创建一个UDF？
浏览 5提问于2017-08-01得票数 1
回答已采纳
1回答
我想扩展pyspark.sql.DataFrame的读写功能，以满足我自己的项目需要。为此，我创建了以下代码
import pyspark.sql
class DataFrame(pyspark.sql.DataFrame):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def write(self, format="parquet", version=True):
if format=="some_format"
浏览 1提问于2020-06-17得票数 0
当我迭代地将500多个列添加到我的pyspark中时，我遇到了堆栈溢出错误。所以我包括了检查点。检查站帮不上忙。因此，我创建了下面的玩具应用程序来测试我的检查点是否正常工作。在这个例子中，我所做的就是一次又一次地复制原始列来迭代地创建列。我坚持，检查点和计数每10个迭代。我注意到我的dataframe.rdd.isCheckpointed()总是返回False。我可以验证检查点文件夹确实是在磁盘上创建和填充的。我在用哥库德的dataproc
这是我的代码：
from pyspark import SparkContext, SparkConf
from pyspark import Stora
浏览 1提问于2017-11-16得票数 1
回答已采纳
5回答
查看新的spark API，还不清楚是否有可能修改DataFrame列。
如何更改数据文件的行x列y中的值？
在pandas中，这将是：
df.ix[x,y] = new_value
编辑：合并下面所说的内容，您不能修改现有的数据格式，因为它是不可变的，但是您可以返回一个新的数据格式，并进行所需的修改。
如果只想根据条件替换列中的值，如np.where
from pyspark.sql import functions as F
update_func = (F.when(F.col('update_col') == replace_val, new_value)
浏览 12提问于2015-03-17得票数 95
回答已采纳
在PySpark中，我们不能使用传统的熊猫或R风格的符号来根据DataFrame中的旧列来制作新的列。例如，假设我试图连接两列：
import pyspark.sql.functions as F
df['newcol'] = F.concat(df['col_1'], df['col_2'])
结果：
TypeError: 'DataFrame' object does not support item assignment
相反，这样做的实现方法是
df = df.withColumn('newcol', F.
浏览 1提问于2016-07-28得票数 1
回答已采纳
这可能是源于我无知的一个愚蠢的问题。我已经在PySpark上工作了几个星期了，一开始我没有多少编程经验。
我的理解是，在Spark中，RDDs、Dataframes和数据集都是不可变的--我同样理解，这意味着您不能更改数据。如果是这样，为什么我们能够使用withColumn()编辑Dataframe的现有列？
浏览 0提问于2018-11-19得票数 14
回答已采纳
我有一个带有n个cols的percentage ( df )，我想要生成另一个n cols的df，其中每一列都记录相应的原始df列中的b/w连续行的百分比差。新df中的列标题应该是旧dataframe + "_diff“中相应的列标题。使用下面的代码，我可以为原始df中的每一列生成百分比变化的新列，但无法将它们插入具有适当列标题的新df中：
from pyspark.sql import SparkSession
from pyspark.sql.window import Window
import pyspark.sql.functions as func
spark = (Sp
浏览 0提问于2018-05-31得票数 0
回答已采纳
我有一个用例来映射基于条件的pyspark列的元素。通过这个文档，我找不到一个函数来执行映射函数。
因此，尝试使用pyspark映射函数，但无法将pyspark列转换为dataFrame
注意:我之所以使用pyspark列，是因为我从我使用的库(远大期望)中获得了它的输入。
@column_condition_partial(engine=SparkDFExecutionEngine)
def _spark(cls, column, ts_formats, **kwargs):
return column.isin([3])
# need to replace the abov
浏览 4提问于2021-11-19得票数 0
回答已采纳
4回答
我有一个具有下面数据(所有列都有字符串数据类型)的Pyspark (原始Dataframe)。在我的用例中，我不确定输入数据中有哪些列。用户只需将dataframe的名称传递给我，并要求我修剪这个dataframe的所有列。典型数据文件中的数据如下所示：
id Value Value1
1 "Text " "Avb"
2 1504 " Test"
3 1
浏览 2提问于2017-07-31得票数 0
我有一个dataframe，它在一个列中存储一个JSON对象。我希望处理JSON对象来创建一个新的dataframe (列的数目和类型不同，每行将从JSON对象生成n个新行)。我在下面编写了这样的逻辑:在迭代原始数据集时，将字典(行)附加到列表中。
data = []
def process_row_data(row):
global data
for item in row.json_object['obj']:
# create a dictionary to represent each row of a new dataframe
浏览 4提问于2022-10-19得票数 1
回答已采纳
1回答
如何计算pyspark中多个列的方差？例如，如果pyspark.sql.dataframe表是：
ID A B C
1 12 15 7
2 6 15 2
3 56 25 25
4 36 12 5
所需的输出为
ID A B C Variance
1 12 15 7 10.9
2 6 15 2 29.6
3 56 25 25 213.6
4 36 12 5 176.2
pyspark中有一个方差函数，但它只能按列工作。
浏览 1提问于2017-04-28得票数 1
1回答
有一段时间，我在寻找如何将多个列一次重命名为一个PySpark DF，并遇到了如下情况：
import pyspark
def rename_sdf(df, mapper={}, **kwargs_mapper):
# Do something
# return something
pyspark.sql.dataframe.DataFrame.rename = rename_sdf
我对最后一段感兴趣，其中通过赋值语句将方法添加到pyspark.DataFrame类中。问题是，我正在创建一个Github来存储我的所有函数和ETL，我认为如果我能够应用上面显示的逻辑，就可以非
浏览 6提问于2020-07-09得票数 1
4回答
PySpark计算相关性
、、、、
我想使用pyspark.mllib.stat.Statistics.corr函数来计算pyspark.sql.dataframe.DataFrame对象的两列之间的相关性。corr函数期望接受rdd of Vectors对象。如何将df['some_name']列转换为rdd of Vectors.dense对象？
浏览 12提问于2016-06-03得票数 16
回答已采纳
我试图将爆炸的列添加到dataframe中：
from pyspark.sql.functions import *
from pyspark.sql.types import *
# Convenience function for turning JSON strings into DataFrames.
def jsonToDataFrame(json, schema=None):
# SparkSessions are available with Spark 2.0+
reader = spark.read
if schema:
reader.schema(s
浏览 9提问于2022-12-02得票数 1
1回答
对于所有想要计算成对行比率的列(例如，行1/行2、行3/行4、行5/行6等)，我都有数据。并将计算结果写入新的数据格式。我决定用3个参数将它封装在一个函数中：
paired_row_rat=function(dataframe,rows,columns){
ratio_df=data.frame(matrix(nrow=rows/2,ncol=columns)) #creates new dataframe
#where number of columns is the same as in dataframe used for
#calculation, number of rows f
浏览 4提问于2017-05-16得票数 0
回答已采纳
2回答
我目前正致力于一个闪闪发光的水应用程序，我是一个完全初学者的火花和h2o。
我想做的是：
加载输入文本文件
创建word2vec模型
使用列word和列向量创建数据格式
将数据作为h2o的输入
通过创建模型，我得到了一个映射，但我不知道如何创建它的数据。输出应该如下所示：
word 向量
断言: 0.3，0.4.
感觉到0.6，0.2.诸若此类。
到目前为止，这是我的代码：
from pyspark import SparkContext
from pyspark.mllib.feature import Word2Vec
from pysparkling import
浏览 4提问于2016-06-28得票数 2
回答已采纳
有没有办法用pyspark 2.1.0创建/填充列，其中列的名称是另一个列的值？我尝试了以下几种方法
def createNewColumnsFromValues(dataFrame, colName, targetColName):
"""
Set value of column colName to targetColName's value
"""
cols = dataFrame.columns
#df = dataFrame.withColumn(f.col(colName), f.col(targetCol
浏览 0提问于2018-09-13得票数 1
我正在使用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
1回答
在pandas中，Pyspark中特定列连接由以下代码执行： datamonthly=datamonthly.merge(df[['application_type','msisdn','periodloan']],how='left',on='msisdn') 我使用Pyspark尝试了类似的逻辑 datamonthly = datamonthly.join(datalabel
, datamonthly['msisdn'] == datalabel['msisd
浏览 13提问于2021-08-30得票数 1
回答已采纳
我想用两个不同的条件和一个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中应用pandas_udf。
我有一个pyspark数据帧，如下所示：
+-------------------+------------------+--------+-------+
| lat| lon|duration|stop_id|
+-------------------+------------------+--------+-------+
| -6.23748779296875| 106.6937255859375| 247| 0|
| -6.2374877929687
浏览 2提问于2020-05-14得票数 2
1回答
我有一个包含多个列的dataframe (在这个问题中，我将使用这个dataframe的一个例子)，并希望创建新的列，当year=2017除以初始列的值时，它将显示特定列的值。更具体地说，以列price为例。第一个新列price_2017只应在year=2017时显示price列的值。下一列price_ratio应该显示等于price/price_2017的值。这些_2017和_ratio后缀被添加到除year之外的所有列中。我知道如何手动操作，但是在实际的dataset中，我有60列，因此需要进行一些优化，比如使用loop或apply，但是不知道如何做。
# importing pandas
浏览 4提问于2022-02-17得票数 1
回答已采纳
我正在尝试理解DataFrame列类型。当然，DataFrame不是一个物化的对象，它只是一组Spark的指令，将来要转换成代码。但我认为，这个类型列表代表了在执行操作时JVM中可能出现的对象类型。
import pyspark
import pyspark.sql.types as T
import pyspark.sql.functions as F
data = [0, 3, 0, 4]
d = {}
d['DenseVector'] = pyspark.ml.linalg.DenseVector(data)
d['old_DenseVector'] =
浏览 1提问于2016-07-31得票数 7
回答已采纳
我试图让pyspark.sql.functions.count()函数或pyspark.sql.functions.count_distinct()函数在计算列中不同元素的数量时考虑null值。
让我给出以下可重复的示例，为此我需要创建一个数据集：
# Dataframe Creation
df = spark.createDataFrame([(1,"arun","engineering",20000),\
(2,"manoj","finance",25000),\
浏览 12提问于2022-06-07得票数 0
我想使用PySpark (Spark1.6.2)对存在于Hive表中的数值数据执行主成分分析(PCA)。我能够将Hive表导入到：
>>> from pyspark.sql import HiveContext
>>> hiveContext = HiveContext(sc)
>>> dataframe = hiveContext.sql("SELECT * FROM my_table")
>>> type(dataframe)
<class 'pyspark.sql.dataframe.D
浏览 1提问于2016-10-06得票数 4
回答已采纳
我有两个PySpark DataFrames，如下所示：
DataFrame A：
+-----+------+
|nodes|counts|
+-----+------+
| [0]| 1|
| [1]| 0|
| [2]| 1|
| [3]| 0|
| [4]| 0|
| [5]| 0|
| [6]| 1|
| [7]| 0|
| [8]| 0|
| [9]| 0|
| [10]| 0|
和DataFrame B：
+----+------+
|nodes|counts|
+----
浏览 2提问于2017-08-16得票数 1
回答已采纳
1回答
我有一个dataframe df，并认为我也可以通过以下操作创建一个新列：
df.Age = 4
这不起作用，因为我只给dataframe一个属性Age，并给它分配数字4。
而且，当我尝试df.head()时，输出也没有显示Age列。
因此，我想知道是否有一种方法可以从dataframe中删除Age，我如何在我的dataframe中创建一个列Age，其中只填充了4的？
浏览 5提问于2014-09-22得票数 1
回答已采纳
在所有发布的关于这个行动的问题中，我找不到有用的东西。
我尝试了几个版本，在所有的版本中，我都有一个DataFrame
dataFrame = spark.read.format("com.mongodb.spark.sql").load()
dataFrame.printSchema()的打印输出是
root
|-- SensorId: string (nullable = true)
|-- _id: struct (nullable = true)
| |-- oid: string (nullable = true)
|-- _type: string (
浏览 1提问于2017-10-25得票数 3
回答已采纳
1回答
我是新来的数据科学，我是使用星火与PySpark的API。我想要创建一个DataFrame文件的.CSV文件。当我这样做时，这些列被移动到一个列中，如下所示。
我用来创建CSV文件的DataFrame的注释是
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("MyFirstCSVLoad").getOrCreate()
df = spark.read.csv("order.csv")
df.Show()
有人能帮我解决这个问题吗。
与CSV文件的链
浏览 3提问于2021-07-04得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
