如何在Pyspark Dataframe中的特定索引中添加行或替换？
在Pyspark中，可以使用union方法来添加行或替换特定索引的行。下面是一个示例：
- 首先，创建一个空的DataFrame作为目标DataFrame，用于存储添加或替换后的结果。
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
target_df = spark.createDataFrame([], schema) # schema为目标DataFrame的结构- 然后，使用
union方法将原始DataFrame中的行添加到目标DataFrame中，除了需要替换的特定索引行。
target_df = target_df.union(original_df.filter(~condition)) # condition为需要替换的特定索引行的条件- 最后，将新的行添加到目标DataFrame中，或者替换特定索引的行。
target_df = target_df.union(new_row) # new_row为需要添加或替换的新行完整的代码示例：
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
# 创建目标DataFrame
target_df = spark.createDataFrame([], schema) # schema为目标DataFrame的结构
# 添加或替换行
target_df = target_df.union(original_df.filter(~condition)) # condition为需要替换的特定索引行的条件
target_df = target_df.union(new_row) # new_row为需要添加或替换的新行在Pyspark中，还可以使用withColumn方法来替换特定索引的行，具体步骤如下：
- 首先，使用
monotonically_increasing_id函数为DataFrame添加一个自增的索引列。
from pyspark.sql.functions import monotonically_increasing_id
original_df = original_df.withColumn("index", monotonically_increasing_id())- 然后，使用
withColumn方法替换特定索引的行。
from pyspark.sql.functions import when
# 使用when函数替换特定索引的行
target_df = original_df.withColumn("column1", when(condition, new_value).otherwise(original_df["column1"]))完整的代码示例：
from pyspark.sql import SparkSession
from pyspark.sql.functions import monotonically_increasing_id, when
spark = SparkSession.builder.getOrCreate()
# 添加自增索引列
original_df = original_df.withColumn("index", monotonically_increasing_id())
# 替换特定索引的行
target_df = original_df.withColumn("column1", when(condition, new_value).otherwise(original_df["column1"]))以上是在Pyspark Dataframe中添加行或替换特定索引的行的方法。请注意，这里的示例代码仅供参考，具体实现可能需要根据实际情况进行调整。
相关·内容
1回答
我如何分配一行与？
、、、
请您将下面这个表达式从Pandas转换为Pyspark，我试着在Pyspark中看到等效的loc吗？
import pandas as pd
df3 = pd.DataFrame(columns=["Devices","months"])
new_entry = {'Devices': 'device1', 'months': 'month1'}
df3.loc[len(df3)] = new_entry
浏览 1提问于2021-02-04得票数 0
回答已采纳
在Pandas dataframe中，我可以使用data.iloc[1:1000,:]获取前1000行。如何在PySpark中做到这一点？
浏览 24提问于2021-01-11得票数 1
回答已采纳
如何修改代码以打印包含功能名称而不仅仅是数字的决策路径。
import pandas as pd
import pyspark.sql.functions as F
from pyspark.ml import Pipeline, Transformer
from pyspark.sql import DataFrame
from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.feature import VectorAssembler
data = pd.DataFrame({
浏览 17提问于2018-08-01得票数 1
回答已采纳
目前，我可以通过RDD使用以下代码将文本文件解析为DataFrame：
def row_parse_function(raw_string_input):
# Do parse logic...
return pyspark.sql.Row(...)
raw_rdd = spark_context.textFile(full_source_path)
# Convert RDD of strings to RDD of pyspark.sql.Row
row_rdd = raw_rdd.map(row_parse_function).filter(bool)
# Conver
浏览 6提问于2016-09-02得票数 0
回答已采纳
我想将此列表L1作为行添加到第一个索引中，如何在Pyspark Dataframe中的特定索引中追加一行？ L1=['na',5.6,2.4]
data=[('fr',8.8,6.6),
('nr',4.4,2.5),
('cc',2.3,3.9)]
data_schema=[StructField('loc',StringType(),True),StructField('col',FloatType(),True),StructField('io',F
浏览 30提问于2021-01-24得票数 0
回答已采纳
我正在使用HashingTF和IDF计算TF和国防军，并且使用以下代码：
from pyspark import SparkContext
from pyspark.mllib.feature import HashingTF
from pyspark.mllib.feature import IDF
sc = SparkContext()
# Load documents (one per line).
documents = sc.textFile("random.txt").map(lambda line: line.split(" "))
hash
浏览 0提问于2016-02-23得票数 0
回答已采纳
我正在尝试将我的excel文件导入Azure-DataBricks机器中的PySpark，我必须将其移动到PySpark Dataframe。我无法执行此操作。获取错误
import pandas
data = pandas.read_excel('/dbfs/FileStore/tables/Time_Payments.xlsx')
df_data = sqlContext.createDataFrame(data)
执行上述操作时，出现以下错误。
Error : field Additional Information: Can not merge type <cla
浏览 2提问于2018-09-18得票数 0
1回答
似乎不支持替换无穷大值。我尝试了下面的代码，但它不起作用。还是我漏掉了什么？
a=sqlContext.createDataFrame([(None, None), (1, np.inf), (None, 2)])
a.replace(np.inf, 10)
或者我必须走一条痛苦的道路:将PySpark DataFrame转换为DataFrame，替换无穷大值，然后将其转换回PySpark DataFrame。
浏览 7提问于2015-12-23得票数 6
回答已采纳
1回答
左反团员
、、、
我有一个dataframe，它有两个列a和b，其中b列中的值是a列中值的a子集。例如：
df
+---+---+
| a| b|
+---+---+
| 1| 2|
| 1| 3|
| 2| 1|
| 3| 2|
+---+---+
我想生成一个包含a和anti_b列的数据格式，其中anti_b列中的值是来自a列的任何值，这样a!=anti_b和行(a,anti_b)就不会出现在原始的数据格式中。因此，在上面的数据中，结果应该是：
anti df
+---+------+
| a|anti_b|
+---+------+
| 3| 1|
| 2| 3|
浏览 2提问于2019-11-18得票数 0
我在我的星火DataFrame上有一篇专栏：
|-- topics_A: array (nullable = true)
| |-- element: string (containsNull = true)
我正在使用CountVectorizer：
topic_vectorizer_A = CountVectorizer(inputCol="topics_A", outputCol="topics_vec_A")
我得到了NullPointerExceptions，因为有时topic_A列包含null。
有办法绕道吗？用零长度数组填充它可以正常工作
浏览 5提问于2016-11-01得票数 4
回答已采纳
我使用pyspark流从kafka获取数据，结果是一个dataframe，当我将dataframe转换为rdd时，它出错了： Traceback (most recent call last):
File "/home/docs/dp_model/dp_algo_platform/dp_algo_core/test/test.py", line 36, in <module>
df = df.rdd.map(lambda x: x.value.split(" ")).toDF()
File "/home/softs/spark-2.4.3-
浏览 56提问于2020-01-06得票数 3
我在我的项目中实现了实验环境。
此组件基于Scikit learn。
在这个组件中，我将给定的CSV读取到pandas数据帧中。在那之后，我选择了最好的特性，并将给定数据帧的维度从100减少到5。之后，我将删除的ID列添加到这个缩减的数据帧中，以供将来使用。通过降维过程删除了这一列。
在我更改代码以读取所有CSV文件并返回一个联合数据帧之前，一切工作正常：
请看下一段代码:读取所有CSV：
dataframes = []
from os import listdir
from os.path import isfile, join
files_names = [f for f in list
浏览 0提问于2016-10-11得票数 1
我正在尝试将pandas.DataFrame代码转换为等效的pyspark DataFrame。我有一个以下格式的RDD。
myRdd = [[1, 'a', {'a':[1, 2]}],
[2, 'b', {'c': 1, 'd':3}],
[3, 'c', {}]]
columnNames = ['sl', 'name', 'params']
内部列表中的第三个元素没有特定的结构。在pandas数据帧中，我可以
浏览 1提问于2018-04-07得票数 1
我希望使用Spark DataFrame的PySpark在特定的单元格中更改一个值。
简单的例子--我创建了一个模拟Spark DataFrame
df = spark.createDataFrame(
[
(1, 1.87, 'new_york'),
(4, 2.76, 'la'),
(6, 3.3, 'boston'),
(8, 4.1, 'detroit'),
(2, 5.70, 'miami'),
(3, 6.320, '
浏览 4提问于2018-05-17得票数 7
回答已采纳
在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数据帧。有没有一种高效的方式在pyspark中做到这一点？？我的datetime列是反转的，所以我需要反转我的dataframe +-------------------+-------+
| date|value_1|
+-------------------+-------+
|2018-11-30 23:59:24| 28.02|
|2018-11-30 23:58:54| 28.02|
|2018-11-30 23:58:24| 28.03|
+-------------------+-------+ 有没有办法在pysp
浏览 33提问于2019-10-18得票数 0
回答已采纳
3回答
我试图用PySpark中的0值替换原始数据中的句号。
我尝试使用.when和.otherwise语句。
我试图使用regexp_replace来更改“.”降至0。
已尝试的代码：
from pyspark.sql import functions as F
#For #1 above:
dataframe2 = dataframe1.withColumn("test_col", F.when(((F.col("test_col") == F.lit(".")), 0).otherwise(F.col("test_col
浏览 8提问于2019-09-01得票数 1
2回答
我已经将我的Crud脚本从本地服务器更新到Internet，它通过任务调度器与批处理文件一起运行，问题是任务调度器不是创建日志文件。
如何使我的服务器使用与任务调度程序一起运行的批处理文件创建日志文件？我已经用脚本创建了批处理文件：
cd E:\www\80\curl
C：\php 5\php E:\www\80\curl\update_database.php >> result.log
但是当我手动运行它时，它确实会创建日志文件。
请帮帮我..。
浏览 6提问于2013-07-17得票数 0
回答已采纳
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
回答已采纳
我是一个用PySpark编程的初学者。我在CSV文件中有以下数据，该文件正在被读取到Spark Dataframe中，并且我想从一个小数据集开始生成一个大型数据集。
# read the csv file in a spark dataframe
df = (spark.read
.option("inferSchema", "true")
.option("header", "true")
.csv(file_path))
我想对每一列中的数据进行混洗，即分别为‘'In
浏览 16提问于2020-05-11得票数 0
1回答
我目前正在阅读Spark the definitive guide，并且有一个通过使用expr来实现DataFrame的示例，但是它不起作用：
from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark.sql import Row
schema = StructType([
StructField("origin", StringType(), True),
StructField("destination", StringType(), True)
浏览 4提问于2020-07-27得票数 0
回答已采纳
2回答
我想把mysql和pyspark连接起来。我正在使用jupyter笔记本来运行pyspark。然而，当我这样做的时候，
dataframe_mysql = sqlContext.read.format("jdbc").options(
url="jdbc:mysql://localhost:3306/playground",
driver = "com.mysql.jdbc.Driver",
dbtable = "play1",
user="root",
password=&
浏览 29提问于2018-08-21得票数 6
我正在尝试使用pyspark读取csv文件，大多数列名都有特殊字符。我想要使用pyspark dataframe.Is删除所有列名中的特殊字符，有什么特定的函数可以同时删除所有列名的特殊字符吗？我很感激你的回应。
浏览 1提问于2020-08-05得票数 2
回答已采纳
我以以下方式创建了一个DataFrame：
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.getOrCreate()
df = spark.read.csv("train.csv", header=True)
我的DataFrame的架构如下：
root
|-- PassengerId: string (nullable = true)
|-- S
浏览 3提问于2018-10-18得票数 0
回答已采纳
1回答
我有一个pyspark管道(包含估算和一个机器学习模型)和一个pandas数据帧。我是否可以在不将其转换为Pyspark dataframe的情况下将管道应用于此pandas数据帧？如果不可能，我如何有效地使用pyspark管道来生成对pandas数据帧的预测？
浏览 3提问于2021-09-13得票数 1
1回答
我试图使用分组Map在PySpark中运行指数加权移动平均。但是，它不起作用：
def ExpMA(myData):
from pyspark.sql.functions import pandas_udf
from pyspark.sql.functions import PandasUDFType
from pyspark.sql import SQLContext
df = myData
group_col = 'Name'
sort_col = 'Date'
schema = df.sele
浏览 0提问于2018-04-30得票数 3
2回答
我遇到过这样的情况:我的列中可能有空值，需要在一个组中求和。 如果我在组中遇到null，我希望该组的和为null。但是默认情况下，Pyspark似乎会忽略null行，并对其余非NULL值求和。 例如： ? dataframe = dataframe.groupBy('dataframe.product', 'dataframe.price') \
.agg(f.sum('price')) 预期输出为： ? 但我得到的是： ? 任何建议都会有帮助！ 提前感谢！
浏览 87提问于2021-01-18得票数 1
回答已采纳
1回答
如何将熊猫DataFrame中的多个值替换为另一个不同的多值(存储在数组中)？
详细说明:我希望替换dataframe 'df‘的每一行名为'Cloud3pm’的值，其中的值为= -1。我希望将这个单元格值分别替换为一个值数组，该数组的长度等于我将在dataframe上找到的值的出现次数等于-1。
浏览 4提问于2022-02-07得票数 -1
回答已采纳
4回答
如果我调用map或mapPartition，并且我的函数接收来自PySpark的行，那么创建本地PySpark或Pandas DataFrame的自然方法是什么？合并行并保留架构的东西？
目前，我所做的事情如下：
def combine(partition):
rows = [x for x in partition]
dfpart = pd.DataFrame(rows,columns=rows[0].keys())
pandafunc(dfpart)
mydf.mapPartition(combine)
浏览 0提问于2015-12-23得票数 2
回答已采纳
我想知道如何使用Python2删除Pyspark Dataframe中的变音符号。我需要像这样的东西
from pyspark.sql.session import SparkSession
from pyspark import SparkContext
import pyspark.sql.functions as sf
from pyspark.sql.types import StringType
df = sc.parallelize([(u'pádlo', 1), (u'dřez', 4)]).toDF(['text', '
浏览 1提问于2019-01-16得票数 0
我有一个非常大的星火DataFrame和许多列，我想作出一个明智的判断是否保持在我的管道中，部分取决于它们有多大。所谓“有多大”，我指的是缓存这个DataFrame时内存中字节的大小，我希望这是对处理这些数据的计算成本的一个不错的估计。有些列是简单类型(例如，双列、整数列)，而另一些列是复杂类型(例如数组和可变长度映射)。
我尝试过的一种方法是在没有列的情况下缓存DataFrame，然后查看Spark中的Storage，然后取不同的内容。但是对于一个列很多的DataFrame来说，这是一个恼人而缓慢的练习。
我通常使用PySpark，所以最好使用PySpark回答，但是Scala也可以。
浏览 0提问于2019-02-25得票数 2
回答已采纳
我正在尝试从pyspark dataframe中获取空值的行。在pandas中，我可以在数据帧上使用isnull()来实现这一点：
df = df[df.isnull().any(axis=1)]
但在PySpark的情况下，当我运行以下命令时，它显示Attributeerror：
df.filter(df.isNull())
AttributeError：'DataFrame‘对象没有属性'isNull’。
如何在不对每一列进行检查的情况下获取具有空值的行？
浏览 25提问于2018-11-27得票数 5
4回答
我有一个具有下面数据(所有列都有字符串数据类型)的Pyspark (原始Dataframe)。在我的用例中，我不确定输入数据中有哪些列。用户只需将dataframe的名称传递给我，并要求我修剪这个dataframe的所有列。典型数据文件中的数据如下所示：
id Value Value1
1 "Text " "Avb"
2 1504 " Test"
3 1
浏览 2提问于2017-07-31得票数 0
我有一个以" null“作为字符串的Dataframe，我想将其替换为数据库中PySpark Dataframe中的SQL值null。 有没有人能帮个忙。我是Spark的新手。 谢谢。
浏览 38提问于2021-09-02得票数 2
我正在使用pandasUDF将标准的ML python库应用于pyspark DataFrame。在定义了模式并进行了预测之后，我得到了pyspark DF作为输出。现在，我想用这个预测数据帧做一些事情，例如，我尝试对列"weekly_forecast_1“中的所有值进行求和。当我应用.collect()或.toPandas()方法时，在.fit()中得到以下错误 IndexError: too many indices for array:array is 0-dimensional, but 1 were indexed 每当我尝试将.collect()或.toPandas()方
浏览 33提问于2020-11-24得票数 0
2回答
我需要使用dataframe来迭代pySpark，就像我们可以使用for循环迭代一组值一样。下面是我写的代码。这段代码的问题是
我必须使用集合来打破并行性。
我无法在函数DataFrame中打印funcRowIter的任何值。
一旦找到匹配项，我就不能中断循环。
我必须在pySpark做这件事，不能用熊猫做这个：
from pyspark.sql.functions import *
from pyspark.sql import HiveContext
from pyspark.sql import functions
from pyspark.sql import Da
浏览 2提问于2017-01-30得票数 3
举个例子，我有一个df
from pyspark.sql import Row
row = Row("v", "x", "y", "z")
df = sc.parallelize([
row("p", 1, 2, 3.0), row("NULL", 3, "NULL", 5.0),
row("NA", None, 6, 7.0), row(float("Nan"), 8, "NULL", float("NaN&
浏览 9提问于2017-12-06得票数 0
回答已采纳
我正在尝试理解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
回答已采纳
2回答
我正试着把下面的熊猫代码翻译成PySpark。但我在这两点上有困难：
但星火DataFrame中有索引吗？
我怎么能像这样在level=0中分组呢？
我在文件里没有发现任何好的东西。如果你有提示的话，我会非常感激的！
df.set_index('var1', inplace=True)
df['varGrouped'] = df.groupby(level=0)['var2'].min()
df.reset_index(inplace=True)
浏览 9提问于2022-10-04得票数 0
回答已采纳
2回答
我在java中有一个类，它构建了一些复杂的星火DataFrame。
package companyX;
class DFBuilder {
public DataFrame build() {
...
return dataframe;
}
}
我将这个类添加到pyspark/木星类路径中，这样它就可以被py4j调用。现在，当我称之为它时，我得到了奇怪的类型：
b = sc._jvm.companyX.DFBuilder()
print(type(b.build()))
#prints: py4j.java_gateway.JavaObject
VS
浏览 0提问于2018-03-14得票数 1
回答已采纳
1回答
我想知道如何用均值替换异常值。我有了dataframe，我可以找到离群值并过滤行，现在我想用平均值替换它。我该怎么做呢？ df类似于： a b
1 27 0
2 10 1
3 80 2
4 21 3
5 46 4
6 100 5 在找到IQR后，我得到的异常值如下所示： Upper = 75
lower = 12
outliers = df.filter((df['a'] > upper) | (df['a'] < lower))
2
浏览 24提问于2021-03-29得票数 0
回答已采纳
我从一个pyspark dataframe开始，在对它使用.take()之后转换成一个列表。我怎么才能保持它是pyspark数据帧呢？
df1 = Ce_clean
print(type(df1))
df1 = df1.take(1000)
print(type(df1))
<class 'pyspark.sql.dataframe.DataFrame'>
<class 'list'>
浏览 1提问于2020-09-17得票数 1
2回答
我有一个pyspark dataframe，其中一些列具有相同的名称。我想将具有相同名称的所有列合并到一列中。例如，输入dataframe： ? 我如何在pyspark中做到这一点？任何帮助都将不胜感激。
浏览 57提问于2021-06-18得票数 2
回答已采纳
我有一个包含两列的数据帧，其中包含数字，我需要按行而不是按列对数据帧进行排序。到处都给出了如何按列对dataframe进行排序，但我找不到如何在pyspark中对dataframe的所有行进行排序。
col1 col2
2 1
3 2
预期输出
col1 col2
1 2
2 3
浏览 0提问于2017-11-14得票数 1
通过执行以下操作，我可以根据数组字段中是否存在特定值来过滤Spark dataframe (在PySpark中)：
from pyspark.sql.functions import array_contains
spark_df.filter(array_contains(spark_df.array_column_name, "value that I want")).show()
有没有一种方法可以获得数组中找到项的位置的索引？它看起来应该存在，但我没有找到它。谢谢。
浏览 1提问于2018-12-13得票数 5
回答已采纳
在python或R中，可以使用索引对DataFrame进行切片。
例如，在熊猫中：
df.iloc[5:10,:]
在pyspark中有没有类似的方式来根据行的位置对数据进行切片？
浏览 3提问于2018-10-13得票数 4
回答已采纳
3回答
`df.loc`的火花源等价？
、、、、
我正在寻找等同于pandas数据帧的pyspark。特别是，我想对pyspark dataframe执行以下操作
# in pandas dataframe, I can do the following operation
# assuming df = pandas dataframe
index = df['column_A'] > 0.0
amount = sum(df.loc[index, 'column_B'] * df.loc[index, 'column_C'])
/ sum(df.loc[index, &
浏览 0提问于2018-05-13得票数 8
回答已采纳
在PySpark中，我使用了dataframe_a：
+-----------+----------------------+
| str1 | array_of_str |
+-----------+----------------------+
| John | [mango, apple] |
| Tom | [mango, orange] |
| Matteo | [apple, banana] |
和dataframe_b
+-----------+----------------------+
浏览 1提问于2020-01-09得票数 1
回答已采纳
我正在尝试将一些pySpark代码转换为Scala，以提高性能。在AWS (它使用Apache )中，会自动为您生成一个脚本，它通常使用DynamicFrame对象加载、转换和写入数据。但是，DynamicFrame类并不具有与DataFrame类相同的所有功能，有时您必须将其转换回DataFrame对象，反之亦然，以执行某些操作。下面是我如何在DataFrame中将DynamicFrame对象转换为pySpark对象：
// PySpark version
// datasource is a DynamicFrame object
datasource0 = dat
浏览 1提问于2018-05-17得票数 3
回答已采纳
1回答
有没有人有一个很好的例子，将Hyperopt集成到星火的MlLib中？我一直试图在Databricks上这样做，并继续获得相同的错误。我不确定这是否与我的目标函数有关，或者它是否与火星之火上的Spark有关，以及它是如何连接到Databricks的。
import itertools
from pyspark.sql import functions as f
from pyspark.sql import DataFrame
from pyspark.sql.types import *
from pyspark.ml import Pipeline, PipelineModel
from
浏览 2提问于2020-02-13得票数 3
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
