通过计算具有整列的行的值，在spark数据帧中创建一列
在Spark数据帧中，可以通过计算具有整列的行的值来创建一列。具体步骤如下：
- 导入必要的Spark库和模块：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col- 创建SparkSession对象：
spark = SparkSession.builder.getOrCreate()- 加载数据集到Spark数据帧：
df = spark.read.csv("data.csv", header=True, inferSchema=True)其中，"data.csv"是数据集的文件路径，header=True表示第一行是列名，inferSchema=True表示自动推断列的数据类型。
- 使用withColumn()方法计算新列的值并添加到数据帧中：
df = df.withColumn("new_column", col("column1") + col("column2"))上述代码中，"new_column"是新列的名称，"column1"和"column2"是用于计算新列值的现有列。
- 显示数据帧的结果：
df.show()这样就在Spark数据帧中成功创建了一列，该列的值是通过计算具有整列的行的值得出的。根据具体的需求，可以使用不同的计算逻辑和函数来创建新列。
推荐的腾讯云相关产品：腾讯云EMR（Elastic MapReduce），是一种大数据处理和分析的云服务，可用于处理Spark数据帧等大数据任务。详情请参考腾讯云EMR产品介绍：腾讯云EMR。
相关·内容
我在Spark 1.3上
我想对数据帧的每一行应用一个函数。此函数对行的每一列进行散列处理，并返回散列列表。
dataframe.map(row => row.toSeq.map(col => col.hashCode))
当我运行这段代码时，我得到了一个NullPointerException。我假设这与有关。
如果不使用嵌套的map，我想不出一种方法来达到同样的效果。
浏览 1提问于2015-12-23得票数 9
回答已采纳
我希望通过将其中一列中的regex值应用于另一列来过滤数据帧。
Example:
Id Column1 RegexColumm
1 Abc A.*
2 Def B.*
3 Ghi G.*
使用RegexColumm过滤数据帧的结果应该是id为1和3的行。
在spark 1.5.1中有没有办法做到这一点？我不想使用UDF，因为这可能会导致可伸缩性问题，寻找spark原生api。
浏览 2提问于2020-03-18得票数 0
我有一个scala函数，它接受一个spark dataframe并返回一个单值，也就是说两个值。这个函数很复杂，使用在DataFrame类中定义的聚合，调用其他java库，并且不能用SQL表达。它需要整个数据帧的内容来进行计算，它不能一次添加一行并建立一个结果。
我有一个大的dataframe，其中包含一个列，我想用它将dataframe分成小块，并对每个小块执行上述计算。然后，我想返回一个新的数据帧，其中包含每个组的一行和两列，一列包含groupby值，另一列包含结果。
使用PandasUDF这将是一个相对简单的任务，但是我不知道如何在Scala中做到这一点。
我尝试使用group by列对
浏览 28提问于2020-04-22得票数 0
scala spark数据帧中有没有df[100, c("column")]的替代品？我想从spark data frame的一列中选择特定的行。例如，上述R等效代码中的100th行
浏览 613提问于2016-02-07得票数 45
回答已采纳
我有一个如下所示的spark数据帧。第一列包含Party文本文件的内容(使用sc.wholeTextfile读取)。第二行包含标识符文本文件的内容。现在，我必须遍历下面数据帧的每一行，并写入不同的hive表。第一行应该写入Party表，第二行应该写入Identifier表，依此类推。
当我使用dataframe.foreaach遍历这个数据帧时，我不能使用spark session，因为spark session是不可序列化的。
+--------------------+--------------+
| content| TblNm|
+------
浏览 2提问于2018-10-06得票数 0
3回答
我有一个Spark Dataframe，其中有一些缺失值。我想通过用该列的平均值替换缺少的值来执行一个简单的补偿。我对Spark非常陌生，所以我一直在努力实现这个逻辑。到目前为止，我做到了以下几点：
A)要对单个列(假设是A列)执行此操作，可以使用下面这行代码：
df.withColumn("new_Col", when($"ColA".isNull, df.select(mean("ColA"))
.first()(0).asInstanceOf[Double])
.otherwise($"ColA"))
b)然而，我还
浏览 1提问于2016-10-15得票数 17
回答已采纳
1回答
我想要一个spark scala代码，它可以减少数据帧大小。有一列是"Status“，还有一列是"Time”。我只想要状态更改时的行。 当前数据帧 Time Status
100 Running
200 Running
300 Stopped
400 Stopped
500 Running
600 Running
700 Stopped
800 Ended 现在我想要一个状态改变时只有行的数据帧。因此，所需的数据帧应为 Time Status
100 Running
300 Stopped
500 Ru
浏览 17提问于2021-07-12得票数 0
回答已采纳
1回答
我希望使用Spark对数据帧的每一列执行Group。Dataframe将有大约。1000列。
我已经尝试对数据帧中的所有列进行迭代，并对每一列执行groupBy。但是程序执行时间超过1.5小时
val df = sqlContext
.read
.format("org.apache.spark.sql.cassandra")
.options(Map( "table" -> "exp", "keyspace" -> "testdata"))
.load(
浏览 0提问于2019-08-12得票数 1
回答已采纳
假设我有一个数据帧：
rdd = sc.textFile(file)
df = sqlContext.createDataFrame(rdd)
df.cache()
我添加了一列
df = df.withColumn('c1', lit(0))
我想重复使用df。那么，我是否需要重新cache()数据帧，或者Spark会自动为我执行此操作？
浏览 14提问于2017-02-05得票数 8
我有一个包含一组计算列的Apache Spark数据帧。对于数据帧中的每一行(大约2000行)，我希望获取10列的行值，并定位第11列相对于其他10列的最接近值。
我想我会把这些行值转换成一个列表，然后使用abs值计算来确定最接近的值。
但是我被困在如何将行值转换为列表的问题上。我获取了一列，并使用collect_list将这些值转换为列表，但不确定当列表来自单行和多列时该如何处理。
浏览 1提问于2017-10-03得票数 0
1回答
我正在尝试通过Spark流读取从kafka传递的行，传入的记录有4列，其中一列有换行符，其中一列用双引号括起来
示例
a ,b , "hi
how are you
fine
",d
那么如何读取或者在通过spark streaming从kafka读取时需要设置哪些选项，以便正确读取记录？
浏览 4提问于2018-08-01得票数 0
<code>B0</code>
我有像图片中这样的数据，其中有两列(Cday，Dday)，但有一些缺失值。不能有两列都有值的行；其中一列或另一列上都有值，或者两者都没有。
我想创建"new“列，它从任何有数字的列中复制值。真的很感谢大家的帮助！
浏览 17提问于2020-03-14得票数 0
我必须创建一个数据框，其中一列填充了runif数字，但一行中的每个数字都比前一个数字小10倍(这应该要做5次)。因此，如果第一行中的第一个数字是5，那么第二个数字是0.5，0.05，0.005，最后是0.0005。例如，在下面的代码中，已经制作了数据帧，我应该像上面解释的那样替换OD列中的前五个值。 levels <- c("ref", "treat", "control")
x <- as.factor(labels)
a <- levels[x]
values <- runif(25,min=0,max=25)
my
浏览 6提问于2019-04-28得票数 0
3回答
我们正在处理无模式的JSON数据，有时spark作业会失败，因为我们在spark SQL中引用的一些列在一天中的某些时间内不可用。在这些小时内，spark作业失败，因为所引用的列在数据帧中不可用。如何处理这种情况？我尝试过UDF，但是我们缺少太多的列，所以不能真正地检查每一列的可用性。我还尝试在更大的数据集上推断模式，并将其应用于数据帧，期望缺少的列将用null填充，但模式应用程序失败并出现奇怪的错误。
请给出建议
浏览 5提问于2018-08-10得票数 2
2回答
我使用的是Java Spark，我有一个这样的数据帧
+---+-----+------+
|id |color|datas |
+----------------+
|1 |blue |data1|
|1 |red |data2|
|1 |orange|data3|
|2 |black |data4|
|2 | |data5|
|2 |yellow| |
|3 |white |data7|
|3 | |data8|
+----------------+
我需要修改这个数据帧，使其看起来像这样：
+---+-------------------
浏览 1提问于2017-05-23得票数 3
我有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
3回答
如果不使用SQL的REPLACE()函数，我们如何才能找到spark数据帧一列中的字数？下面是我正在使用的代码和输入，但replace()函数不起作用。
from pyspark.sql import SparkSession
my_spark = SparkSession \
.builder \
.appName("Python Spark SQL example") \
.enableHiveSupport() \
.getOrCreate()
parqFileName = 'gs://caserta-pyspark-eval/t
浏览 3提问于2018-02-22得票数 10
1回答
我正在对数据帧中的一列应用lag函数。应用lag函数的列已包含一些空值。如何排除对有NULL值的行计算lag？ 以下是来自dataframe的数据： +------------+-------------+-----+-----+
| NUM_ID | TIME |SG1_V|SG2_V|
+------------+-------------+-----+-----+
|XXXXX01 |1570167499000| |79.0 |
|XXXXX01 |1570167502000| |88.0 |
|XXXXX01 |15701
浏览 30提问于2019-10-14得票数 1
2回答
我试着理解pyspark.sql.currenttimestamp()和datetime.now()之间的行为差异
如果我使用这2种机制来创建时间戳列，在DataBricks中创建一个Spark，那么一切都像预期的那样工作得很好.
curDate2 = spark.range(10)\
.withColumn("current_date_lit",F.lit(date.today()))\
.withColumn("current_timestamp_lit",F.lit(F.current_timestamp()))\
.withColumn(&#
浏览 9提问于2022-02-12得票数 2
回答已采纳
我正在运行火花-提交在我的库伯内特斯集群。根据星星之火文档，我正在创建一个pod模板，并指定重新启动策略Always。下面是我的吊舱模板。
apiVersion: v1
kind: Deployment
metadata:
name: pi-exampe
spec:
restartPolicy: Always
Command:
/usr/local/lib/python3.9/site-packages/pyspark/bin/spark-submit --class org.apache.spark.examples.SparkPi --master k8s://https://k
浏览 2提问于2021-11-01得票数 0
4回答
如何检查两个对象(例如数据帧)在R中的值是否相等？
所谓值相等，是指一个数据帧的每一列的每一行的值等于第二个数据帧中相应行和列的值。
浏览 3提问于2012-05-15得票数 57
回答已采纳
1回答
我需要了解如何在Spark中创建任意大小的DenseVectors的单位矩阵。我试图从mllib.linalg.distributed模块做一些事情，但是没有用。我需要的是一个只有一列“特征”的数据帧，它以DenseVectors作为它的行，其中每一行都是单位矩阵中的对应行。
浏览 19提问于2017-02-09得票数 1
回答已采纳
我有一个spark数据帧，只想删除最后一列。
我试过了
df.drop(df.columns.last)`
但是得到了错误AttributeError: 'list' object has no attribute 'last'。
我也试过了：
df = df.drop(df.columns[-1])
但这会删除与last同名的所有列。
使用Spark 2.4
浏览 0提问于2020-01-24得票数 0
1回答
我有一个具有10^8行数的spark数据帧df。我已经在该数据帧上添加了一列，作为rowId，我希望将其用作主键。我使用下面的命令做了同样的事情
df.withColumn("rowId"，monotonically_increasing_id())
现在，我想从该数据帧中选择一个新的数据帧，其中包含一些选定的行数，我已经知道这些行的索引是以列表的形式存在的。如果有人能帮助我用列表中选定的行数形成一个新的数据框，那将对我很有帮助。
浏览 0提问于2017-10-28得票数 0
我们是否可以使用Spark-SQL或scala检查spark数据帧中的每一列是否都包含特定的字符串(例如"Y")？
我已经尝试了以下方法，但我认为它不能正常工作。
df.select(df.col("*")).filter("'*' =='Y'")
谢谢，Sai
浏览 0提问于2017-09-09得票数 2
回答已采纳
1回答
在Spark环境中混合使用python map和lambda函数时，我遇到了一个问题。
给定df1，我的源数据帧：
Animals | Food | Home
----------------------------------
Monkey | Banana | Jungle
Dog | Meat | Garden
Cat | Fish | House
Elephant | Banana | Jungle
Lion | Meat | Desert
我想创建另一个数据帧df
浏览 0提问于2019-06-24得票数 1
回答已采纳
我想为一个大的spark数据帧计算成对的kendall的tau等级相关性。它很大(比如10m行，10k列)，不能转换成pandas数据帧，然后使用pandas.DataFrame.corr进行计算。
此外，每列可能具有空值，因此在计算成对的kendall's tau时，需要排除两列中任何一列中具有空值的行。
我查过pyspark.mllib.stat.Statistics.corr了。它支持"pearson“和"spearman”。
df_rdd = df.rdd.map(lambda row: row[0:])
corr_mat = Statistic
浏览 18提问于2019-07-20得票数 2
2回答
我有一个excel文件，其中包含一列和多行。每一行包含1个邮政编码(或邮政编码)。我想要在地图上每一行显示一个小点，这取决于它在地图上的位置。
浏览 3提问于2016-07-05得票数 0
2回答
我有6个csv文件，其中一列是句子，第二列是整数。
所有csv文件中的句子都是相同的，但它们在文件之间的关键顺序是不同的。
我想逐句合并所有数据帧，这样我就有了一列句子，然后每个整数列都与来自每个csv文件的句子相关联。
我尝试了常见的“句子”列的各种合并和缩减技术，但最终得到的行数比正常情况下多了几个数量级。
例如：
data_frames = [df1, df2, df3, df4, df5, df6]
reduce(lambda x,y: pd.merge(x,y, on='sentence', how='inner'), data_frames)
产生一
浏览 0提问于2020-01-07得票数 1
我正在尝试使用scala中的withColumn向spark数据帧中添加一列，
因为数据集很大，所以我的withColumn在特定的记录上失败了，
我正在尝试用try-catch编写这段代码，这样我就可以捕获异常并记录这些错误记录，然后继续剩余的记录。
在spark-scala中有没有一种方法来处理withColumn中发生的异常？
请提个建议。
浏览 2提问于2020-03-04得票数 0
1回答
我正在做一个使用spark的项目。在某些阶段，我需要在单个数据帧中合并或连接3个数据帧。这些数据帧来自spark sql表，我使用了联合函数，它已经合并了两个表中具有相同编号的列，但我也需要合并不相等的列值。我现在很困惑，有没有办法在pyspark中合并或连接不相等的基于列的数据帧，请指导我
浏览 13提问于2016-09-22得票数 0
我在R中有一个5列的数据帧。第一列包含行名，但R将其识别为变量，因此这将在后续步骤中抛出错误。关于如何让R识别这些是行名而不是变量，有什么想法吗？非常感谢大家的帮助！
**我正在做一些基因测序分析，数据框包含临床信息。因此，我更喜欢保持数据帧的完整性，因为行名包含用于跟踪单个样本的标识符。
浏览 40提问于2017-01-26得票数 0
1回答
从这里开始：
items.take(2)
[['home', 'alone', 'apparently'], ['st','louis','plant','close','die','old','age','workers','making','cars','since','onset','mass','automotive'
浏览 6提问于2021-03-06得票数 0
回答已采纳
1回答
火花数据帧
、、
星火数据帧和Pandas数据帧的区别是什么？我的理解是，Pandas数据框架主要用于将CSV数据读入DF，在DF中，Spark数据帧用于将弹性分布式数据( RDD )加载到数据帧中，并且可以操作数据帧。请分享你的反馈
浏览 6提问于2020-09-01得票数 0
回答已采纳
我有一个数据帧，其中每一列都由0和另一个数字组成。例如： ? 我想要操作数据帧，以便包含相同其他数字的列成为一列，其中如果另一个数字出现在每行中，则该值与另一个数字相同，否则它将变为零。 例如，我希望上面的数据帧看起来像这样 ..1 ..2 ..3
1 2 3
0 2 0
0 0 0
1 0 0 数据帧的第一行是1，因为原始数据帧的第一行中的值都是1。第一列的第二行是0，因为该行中有1和0。 以下是一些可重现的数据： structure(list(...1 = c(1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0,
浏览 9提问于2021-01-20得票数 0
回答已采纳
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回答
我有一个用逗号分隔的csv文件，每行有215列。文件非常大，我使用pandas dataframe pd.read_csv()和块来读取数据。文件中的一行有214列，而不是215列，但dataframe读取数据时将NaN应用于最后一列。 有没有办法在pandas dataframe中标识此行，因为我更喜欢dataframe而不是csv_reader。 注意:最后一列也可以为null，因此按等于null的最后一列过滤数据帧将不起作用
浏览 19提问于2020-09-30得票数 0
下面是动态创建的Spark
#^name#^name#^name#.
+
安迪(20，10，30)
\x{e76f}(30，40，20)
Joe (40，90，60)
+
现在，我需要迭代Spark中的每一行和每一列来打印以下输出，如何做到这一点？
安迪20
安迪10
安迪30
Berta 30
Berta 40
Berta 20
乔40
乔90
乔60
浏览 1提问于2019-05-29得票数 0
回答已采纳
1回答
我有一个使用Java的Spark大型项目。我读取超过1.000.000行的csv文件，其中一列是字符串。
当我尝试执行一个VectorAssembler来使用ML算法时，我有一个错误，因为“月亮”列是一个字符串。
因此，我试图用以下方式将这个字符串转换为Integer：
Dataset<Row> moons = typedMoons.withColumn("Moon", typedMoons.col("Moon").cast("Integer"));
但是当我这样做的时候，我得到了那个列中的一个空值。
因此，我试图用na.fill()
浏览 0提问于2018-09-28得票数 0
回答已采纳
如果我不缓存使用带有limit选项的spark SQL生成的dataframe，当我编辑得到的dataframe并显示它时，我会得到不稳定的结果吗？ 描述。 我有一个类似下面的表格，它是通过使用带有limit选项的spark SQL生成的： +---------+---+---+---+---+
|partition| | 0| 1| 2|
+---------+---+---+---+---+
| 0| 0| 0| 10| 18|
| 1| 0| 0| 10| 17|
| 2| 0| 0| 13| 17|
+---------
浏览 6提问于2021-04-21得票数 0
回答已采纳
1回答
我在代码中发现了一个bug，其中一个数据帧被分割成比预期更多的分区(超过700个)，当我试图将它们重新分区到只有48个时，这会导致太多的随机操作。我不能在这里使用coalesce()，因为在进行重新分区之前，我希望首先有更少的分区。
我正在寻找减少分区数量的方法。假设我有一个spark数据帧(包含多个列)划分为10个分区。我需要基于其中一列进行orderBy转换。此操作完成后，生成的数据帧是否具有相同数量的分区？如果没有，spark将如何决定分区的数量？
另外，除了像repartition()这样明显的转换之外，我还需要注意哪些其他转换可能会导致数据帧的分区数量发生变化？
浏览 2提问于2017-09-13得票数 1
spark是个新手，我边学边学。我有一个非常大的文本文件，其中的列由“|”分隔，我想将其插入到spark dataframe中。但是，该文件只有一个行字符串。该文件如下所示：
col1|||||col2|||||col3|||||col4|||||col5|||||col1|||||col2|||||col3...
因此，第1列到第5列基本上只在一行中循环。我尝试使用sed命令在每隔5个“|”后插入一行，方法是：
sed -r 's/([^|||||]*|||||){5}/&\n/g'
它在很大程度上起到了作用，但最终由于某种原因而无法正常工作。我怀疑col4 (这是
浏览 0提问于2018-10-08得票数 0
我有一个包含一百万行和560列的spark数据帧。我需要找到dataframe的每一列中唯一项的计数。我已经编写了以下代码来实现这一点，但它被卡住了，并且执行起来花费了太多的时间：
count_unique_items=[]
for j in range(len(cat_col)):
var=cat_col[j]
count_unique_items.append(data.select(var).distinct().rdd.map(lambda r:r[0]).count())
cat_col包含所有分类变量的列名
有什么方法可以优化这一点吗？
浏览 9提问于2016-11-29得票数 0
我被要求在apache spark sql (java api)中通过dataframes做一些事情，我认为如果按照一种简单的方法执行，将会花费很大的成本(我仍在使用简单的方法，但我认为它将花费很大，因为它至少需要4种连接)。
我得到了以下数据帧：
+----+----+----+----+----+----------+------+
| C1| C2| C3| C4| C5|UNIQUE KEY|points|
+----+----+----+----+----+----------+------+
| A| A|null|null|null| 1234|
浏览 10提问于2020-03-10得票数 2
回答已采纳
我需要向spark数据帧添加一列，该列应该是重复的序列号，例如[1, 1, 1, 2, 2, 2, 3, 3, 3, ..., 10000, 10000, 10000]。我知道我们可以使用monotonically_increasing_id来获得作为新列的序列号。 val df_new = df.withColumn("id", monotonically_increasing_id) 那么，扩展此函数以获取重复序列号的解决方案是什么？谢谢!
浏览 20提问于2021-03-26得票数 0
回答已采纳
2回答
有一个数据帧 data = [(42, """[{"param_a":9519,"param_b":7,"param_c":64},{"param_a":7483,"param_b":7,"param_c":1},{"param_a":1032,"param_b":7,"param_c":0}]""")]
df = spark.createDataFrame(data, ['key',
浏览 72提问于2020-12-18得票数 1
回答已采纳
1回答
我知道我可以“引爆”一列类型数组，如下所示：
import org.apache.spark.sql._
import org.apache.spark.sql.functions.explode
val explodedDf =
payloadLegsDf.withColumn("legs", explode(payloadLegsDf.col("legs")))
现在我有多个行；数组中的每个项都有一个行。
有什么办法我可以“用指数爆炸”吗？这样就会有一个新列，其中包含原始数组中项的索引？
)我可以想出办法来做这件事。首先，将数组字段设置为原始值和索
浏览 0提问于2018-06-21得票数 11
回答已采纳
我在这里使用PySpark数据帧。"test1“是我的PySpark数据帧和事件 _ date是一个TimestampType。因此，当我尝试获取事件的不同计数时 _ date，结果是一个整数变量，但是当我尝试获取同一列的max时，结果是一个dataframe。我想了解哪些操作会导致数据帧和变量。我还想知道如何将事件日期的最大值存储为变量 生成整数类型的代码： loop_cnt=test1.select('event_date').distinct().count()
type(loop_cnt) 生成数据帧类型的代码： last_processed_dt=test1
浏览 147提问于2016-05-03得票数 5
回答已采纳
2回答
这是的后续问题。我想要检查的是，数据框中的任何一列是否包含所有行的相同值(数值或字符串)。例如,
sample <- data.frame(col1=c(1, 1, 1), col2=c("a", "a", "a"), col3=c(12, 15, 22))
其目的是检查数据帧中的每一列，以查看对于所有行，哪一列不具有相同的条目。该怎么做呢？特别是，既有数字也有字符串。
我的预期输出应该是一个包含列号的向量，该列号具有不同的条目。
浏览 0提问于2018-08-27得票数 3
我有两个数据帧，我需要使用具有两个连接谓词的非相等连接(即不等式连接)将它们连接在一起。
一个数据帧是直方图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
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
