pyspark dataframe错误：_()缺少1个必需的位置参数：'col‘
pyspark dataframe错误：_()缺少1个必需的位置参数：'col'
这个错误是由于在使用pyspark dataframe时，调用了一个函数_()，但是没有传入必需的位置参数'col'导致的。
解决这个错误的方法是确保在调用_()函数时传入正确的位置参数'col'。根据错误信息，'col'可能是指代列名或者某个列的引用。
以下是一种可能的解决方法：
- 确保你已经正确导入了pyspark模块和相关的函数。
from pyspark.sql import SparkSession
from pyspark.sql.functions import _- 确保你已经创建了SparkSession对象，并加载了需要处理的数据。
spark = SparkSession.builder.appName("example").getOrCreate()
df = spark.read.csv("data.csv", header=True)- 在使用_()函数时，确保传入了正确的位置参数'col'。例如，假设你想对某一列进行操作，可以使用以下代码：
df = df.withColumn("new_col", _(col("old_col")))在这个例子中，_()函数用于对"old_col"列进行操作，并将结果存储在"new_col"列中。
- 如果你需要进一步处理DataFrame，可以继续使用其他pyspark dataframe的函数和操作。
这是一个简单的解决方案示例，具体的解决方法可能因实际情况而异。如果你需要更多关于pyspark dataframe的信息，可以参考腾讯云的相关产品文档和教程：
相关·内容
我正在运行一个Spark作业，它从Teradata DBMS读取一个带有SQL查询的DataFrame。 当作业将S3上的文件写为parquet时， partition_keys = ["Cat$col1", "Cat$col2"]
df.write.mode("overwrite").partitionBy(partition_keys) 抛出以下java.lang.ClassCastException异常： File "/lib/python3.7/site-packages/pyspark/python/lib/pyspark.z
浏览 27提问于2021-07-15得票数 0
1回答
多列上的多聚合
、、
我使用Python在Pyspark框架中。我试图使用groupby在不同的列上应用不同的聚合。
我有一个包含col1、col2、col3、col4列的df，我想做这样的事情：df.groupby("col1").sum("col2", "col3").avg("col4")
但是我发现了一个错误：
"/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py"，'DataFrame‘对象没有属性'avg’回溯(最近一次调用)：文件第1
浏览 4提问于2019-09-27得票数 0
回答已采纳
我试图在Dataproc集群上运行一些python pyspark脚本，但失败了，错误如下：
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py", line 815, in join
if isinstance(on[0], basestring):
IndexError: list index out of range
我在代码中使用的语法是：-
df1.join(df2, col1)
有什么想法吗？
浏览 0提问于2018-01-05得票数 0
1回答
当我尝试运行下面的代码时，我得到了一个错误: TypeError: splitter()得到了一个意外的关键字参数'*argv‘。
我想知道是否有更合适的方法来添加这种类型的参数？这个参数的目的是允许使用函数的人为*argv添加0-inf参数，我知道它正在以我想要的方式工作。我只是不知道如何解析这个参数。
import pandas as pd
import numpy as np
import argparse
def arg_parse():
parser = argparse.ArgumentParser()
parser.add_argument("
浏览 3提问于2020-05-28得票数 0
目的:实现此查询 select *,
case when new_x != x or new_y != y then 'some_status_change' else cdc_status end as cdc_status
from dataframe where cdc_status = 'noUpdateRequired' 我正在尝试使用pyspark(3.0.0)和spark(2.4.4)来实现这个逻辑，我现在有这个 df = df.withColumn("cdc_status",
浏览 100提问于2021-05-06得票数 0
回答已采纳
我正在尝试通过SparkNLP对文本数据进行主题建模和情感分析。我已经对数据集执行了所有预处理步骤，但在LDA中遇到错误。 Error 程序是： from pyspark.ml import Pipeline
from pyspark.ml.feature import StopWordsRemover, CountVectorizer, IDF
from pyspark.ml.clustering import LDA
from pyspark.sql.functions import col, lit, concat, regexp_replace
from pyspark.sql.ut
浏览 187提问于2021-04-22得票数 1
回答已采纳
有没有办法用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
下面的代码可以从pyspark解释器中运行。
spark_utils.py
--------------
from typing import List, Optional
from pyspark.sql import DataFrame
from pyspark.sql import functions as F
def add_new_cols_from_list(
df: DataFrame, columns: List[str], default_value: Optional[F.column] = F.lit(0.0)
) -> DataFrame:
re
浏览 5提问于2022-01-25得票数 0
1回答
我正在尝试将hql脚本转换为pyspark。我正在努力在groupby子句之后的聚合中实现case when语句的总和。例如：
dataframe1 = dataframe0.groupby(col0).agg(
SUM(f.when((col1 == 'ABC' | col2 == 'XYZ'), 1).otherwise(0)))
在pyspark中是可能的吗？我在执行这样的语句时遇到错误。谢谢
浏览 0提问于2016-11-23得票数 1
回答已采纳
1回答
如何删除“。”从星火DataFrame的列名？
重命名具有“”的列名的DataFrame.select(F.col().alias())方法。在他们身上抛出一个错误。
下面的代码是可复制的。
# import Spark libraries, configuration, Contexts, and types.
import pyspark
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
#############
# Start Spark.
spark = SparkSession.builde
浏览 12提问于2017-11-15得票数 3
回答已采纳
我有一个带有2列(col1和col2)的pyspark中的dataframe，col2是一个行列表(dataframe在col1上减少了)。现在，我想使用neo4j将此数据写入py2neo。如何编写和格式化密码查询字符串？如果我要使用火花连接器编写数据，我的查询如下-
query_sparkneo4j_connector = "MERGE (d:Node1 {Node1: event.col1}) \
FOREACH (i in event.col2 | \
CREATE (c:Node2 {Prop1: i.xx, Prop2: i.yy})
浏览 7提问于2022-06-28得票数 0
我正在用PySpark 3.2.1开发一个程序。
Mypy == 0.950
其中一个操作需要将小DataFrame的信息转换为列表。
守则是：
result = df.select("col1","col2","col3").toPandas().values.tolist()
我需要把它转换成一个列表，因为我随后广播了信息，而pyspark广播不能是DataFrame
对于这段代码，我得到以下mypy错误：
error: "DataFrameLike" has no attribute "values"
我可能
浏览 4提问于2022-05-20得票数 -1
1回答
获得了以下pyspark代码：
import pyspark.sql.functions as F
null_or_unknown_count = df.sample(0.01).filter(
F.col('env').isNull() | (F.col('env') == 'Unknown')
).count()
在测试代码中，数据帧是模拟的，所以我尝试为这个调用设置return_value，如下所示：
from unittest import mock
from unittest.mock import ANY
@mock.pa
浏览 11提问于2019-11-02得票数 12
回答已采纳
在pyspark中，有没有一种方法可以在DataFrame列上执行substr函数，而不指定长度？也就是说，像df["my-col"].substr(begin)这样的东西。
浏览 0提问于2019-09-10得票数 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
回答已采纳
我有一个电火花数据，我想删除任何一行至少有一个NA。
我知道如何只对一列(下面的代码)这样做。
如何对dataframe的所有列执行相同的操作？
可复制示例
# Import modules
from pyspark.sql import SparkSession
from pyspark import SparkContext
from pyspark.sql.functions import col
from pyspark.sql import Row
# Defining SparkContext
SparkContext.getOrCreate()
# Defining Spa
浏览 14提问于2022-10-05得票数 1
回答已采纳
我想通过pyspark在spark dataframe的一个向量列中找到最大值的索引。 我的火花是 3.0.0 df： id val (vector (nullable = true))
516 0: 1 1: 10 2: [] 3:[0.162, 0.511, 0.022, ....] 这是一个稀疏向量吗？如何访问数组？ [0.162, 0.511, 0.022, ....] 基于How to find the index of the maximum value in a vector column?、How to get the index of the h
浏览 8提问于2020-09-22得票数 3
我试图从原始的Dataframe "User_df“中获取每个列的列名和数据类型，到它们自己的列中，其中列的值分别是列名和数据类型。
我有88列和9500行。
当我尝试用一个列("ColumnDataTypes")创建一个Dataframe时，下面的代码是有效的，但当我试图为这两个列创建一个Dataframe时，代码就不起作用了：
col_types = User_df.dtypes
col_types2, col_types3 = [items[:] for items in zip(*col_types)]
rdd_col_names = sc.parallel
浏览 36提问于2022-08-16得票数 0
回答已采纳
1回答
我有一个，它有10列从一个拼花文件中读取。
df = spark.read.parquet(path)
我想将几个预处理步骤应用到这个dataframe列的一个子集：col_list。
下面的工作很好，但除了有点丑，我也有感觉它不是最佳的。
import pyspark.sql.functions as F
for col in col_list:
df = df.withColumn(col, F.regexp_replace(col, ".", " ")
df = df.withColumn(col, F.regexp_replace(co
浏览 7提问于2021-06-14得票数 0
回答已采纳
在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
回答已采纳
我已经编辑了这个问题来提供一个例子-
我有一个列名的列表：
colnames = ['col1','col2','col3']
我需要一个接一个地将它们传递给Dataframe函数，以返回每个函数的值。我不会使用groupBy函数，因此这不是另一个问题的重复。我只需要Dataframe中每列所有值的最大值、最小值、和。
代码示例：
from pyspark import SparkContext
from pyspark.sql import SQLContext
sc = SparkContext("local[2]", &#
浏览 0提问于2018-05-24得票数 1
5回答
我正在开发一个包含n列的PySpark DataFrame。我有一组m列(m < n)，我的任务是选择其中包含最大值的列。
例如：
输入: PySpark DataFrame，包含：
col_1 = [1,2,3], col_2 = [2,1,4], col_3 = [3,2,5]
输出：
col_4 = max(col1, col_2, col_3) = [3,2,5]
正如问题中所解释的那样，熊猫身上也有类似的东西。
在PySpark中是否存在这样的方法，或者我是否应该将PySpark df转换为Pandas，然后执行这些操作？
浏览 7提问于2016-06-07得票数 29
回答已采纳
首先，我要预先感谢大家的帮助！我有4张桌子，我加入了它们，得到了一个PySpark数据。其中一个dataframe列如下所示，它有大约20万条记录：
{"table_name":"BTR.DAILY_BTR.JSC_MON","login":"0015471"}
{"table_name":"BTR.DAILY_BTR.ESHOP.JSC_MON","login":"0015471"}
该列的类型为“string”。我需要通过key table_name获得价值。我尝
浏览 4提问于2022-04-26得票数 0
from pyspark.sql.window import Window
import mpu
from pyspark.sql.functions import udf
from pyspark.sql.functions import lag
from math import sin, cos, sqrt, atan2
windowSpec = Window.partitionBy("UserID").orderBy(asc("Timestamp"))
df14=df.withColumn("newLatitude
浏览 3提问于2022-01-09得票数 0
我有一个很大的pyspark.sql.dataframe.DataFrame，我想保留(所以filter)保存在location列中的所有行，其中的URL包含预先确定的字符串，例如'google.com‘。
我试过了：
import pyspark.sql.functions as sf
df.filter(sf.col('location').contains('google.com')).show(5)
但是这抛出了一个
TypeError: _TypeError: 'Column' object is not callable
浏览 2提问于2017-01-27得票数 66
回答已采纳
2回答
我在pyspark中有一个非常简单的dataframe，类似于： from pyspark.sql import Row
from pyspark.mllib.linalg import DenseVector
row = Row("a", "b")
df = spark.sparkContext.parallelize([
offer_row(DenseVector([1, 1, 1]), DenseVector([1, 0, 0])),
]).toDF() 我想要计算这些向量的点积，而不是求助于UDF调用。 spark MLLIB documen
浏览 17提问于2019-05-02得票数 1
1回答
我有一份数据文件，在Pyspark as中：
listA = [(1,'AAA','USA'),(2,'XXX','CHN'),(3,'KKK','USA'),(4,'PPP','USA'),(5,'EEE','USA'),(5,'HHH','THA')]
df = spark.createDataFrame(listA, ['id', 'name','countr
浏览 0提问于2018-12-05得票数 2
回答已采纳
使用带有数据的when时出现的when错误
pySpark尝试
test = filterDomainLookup.when((col("PrimaryLookupAttributeName") == "") & (col("SecondaryLookupAttributeName") == ""), "Not Defined")
display(test)
AttributeError: 'DataFrame' object has no attribute 'when'
浏览 1提问于2020-05-27得票数 0
回答已采纳
1回答
我有一个带有日期列和整数列的dataframe，我想根据整数列将月份添加到date列中。我尝试了以下方法，但是我得到了一个错误：
from pyspark.sql import functions as f
withColumn('future', f.add_months('cohort', col('period')))
其中'cohort‘是我的date列，period是一个整数。我得到了以下错误：
TypeError:列不可迭代
浏览 2提问于2022-02-08得票数 0
回答已采纳
使用snowflake spark连接器从snowflake复制表时，默认行为是将结构化数据映射到spark字符串：https://docs.snowflake.net/manuals/user-guide/spark-connector-use.html#from-snowflake-to-spark-sql 例如，给定snowflake中的一个表： create table schema.table as
select
array_construct('1','a') as array_col,
object_construct(
浏览 26提问于2019-05-30得票数 1
我有一个pyspark格式的dataframe，读起来如下： df = spark.table('db.table')
.select(F.col('key').alias('key_a'),
F.to_date('move_out_date', 'yyyyMMdd').alias('move_out_date')) 现在，我想将move_out_date列与20151231日期进行比较。但是下面的代码不能正常工作 from pysp
浏览 17提问于2019-08-31得票数 1
回答已采纳
当将Kafka消息转换为dataframe时，在将包作为参数传递时会出现错误。
from pyspark.sql import SparkSession, Row
from pyspark.context import SparkContext
from kafka import KafkaConsumer
import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0.jar: org.apache.spark:spa
浏览 1提问于2020-06-12得票数 2
回答已采纳
2回答
比较两个中的字符串值
、、、、
我有两个名为- brand_name和poi_name的数据格式。
Dataframe 1(brand_name):-
+-------------+
|brand_stop[0]|
+-------------+
|TOASTMASTERS |
|USBORNE |
|ARBONNE |
|USBORNE |
|ARBONNE |
|ACADEMY |
|ARBONNE |
|USBORNE |
|USBORNE |
|PILLAR |
+-------------+
Dataframe 2:-(po
浏览 0提问于2018-05-23得票数 1
回答已采纳
1回答
在Pyspark中，我需要按ID分组并创建四个新列(min，max，std，ave)。 from pyspark.sql import functions as F
from pyspark.sql.window import Window
w = (Window.orderBy(F.col("Date").cast('long')).rowsBetween(-4, 0))
df = df.groupby("ID") \
.withColumn('hr1_ave', F.avg("rpm").over(w))
浏览 23提问于2020-09-22得票数 1
回答已采纳
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
1回答
我正在使用一个自定义函数作为reduce操作的一部分。对于下面的例子，我得到了下面的消息TypeError: reduce() takes no keyword arguments -我相信这是由于我在函数exposed_colum中使用字典mapping的方式-你能帮我修复这个函数吗？
from pyspark.sql import DataFrame, Row
from pyspark.sql.functions import col
from pyspark.sql import SparkSession
from functools import reduce
def proces
浏览 18提问于2020-02-05得票数 0
回答已采纳
1回答
我有PK列的列表。我在存储主键，因为每个表的主键数可能会发生变化。我想根据pk_list中的列连接两个数据帧。
pk_list=['col1',col2', .... 'coln']
现在，我的代码如下所示：
full_load_tbl_nc = full_load_tbl.join(delta_load_tbl, (col(f) == col(s) for (f,s) in zip(pk_list,pk_list) ) , "leftanti")
当我运行代码时，我会得到以下错误：
在join "/mnt/yarn/userc
浏览 0提问于2018-12-07得票数 2
我有一个PySpark DataFrame，我尝试过许多示例，演示如何使用现有列创建一个基于操作的新列，但它们似乎都不起作用。
因此，我有一个问题：̶w̶o̶：
1-为什么这段代码不能工作？
from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
import pyspark.sql.functions as F
sc = SparkContext()
sqlContext = SQLContext(sc)
a = sqlContext.createDataFrame([(5, 5, 3)
浏览 0提问于2017-06-08得票数 0
回答已采纳
当我试图连接两只熊猫数据时，我发现了下面的一个错误：
TypeError: cannot concatenate object of type 'list; only ps.Series and ps.DataFrame are valid
一开始，我认为这是因为一种包含某列列表的数据格式。因此，我尝试将两个不包括列中列表的数据文件连接起来。但我也犯了同样的错误。为了确定，我打印了数据格式类型。他们都是pandas.core.frame.DataFrame。为什么我会有这个错误，即使它们不是列表？
import pyspark.pandas as ps
split_col = spli
浏览 8提问于2022-11-28得票数 0
回答已采纳
1回答
我正在尝试创建一个包含日期范围的单一列的PySpark数据框架，但是我一直收到这个错误。我也尝试将它转换为int，但我不确定您是否应该这样做。
# Gets an existing SparkSession or, if there is no existing one, creates a new one
spark = SparkSession.builder.appName('pyspark-shellTest2').getOrCreate()
from pyspark.sql.functions import col, to_date, asc
from pyspar
浏览 20提问于2022-12-02得票数 0
我有以下DataFrame：
+----------+
| date|
+----------+
|2017-01-25|
|2017-01-21|
|2017-01-12|
+----------+
下面是在DataFrame上面创建的代码：
import pyspark.sql.functions as f
rdd = sc.parallelize([("2017/11/25",), ("2017/12/21",), ("2017/09/12",)])
df = sqlContext.createDataFrame(rdd, [&#
浏览 4提问于2018-01-20得票数 10
回答已采纳
我想使用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
回答已采纳
1回答
我对pyspark是个新手，正在做我的第一个spark项目，我面临着两个问题。
a)无法使用以下方法引用列
df["col1"].show()
***TypeError: 'Column' object is not callable***
b)不能将我的spark数据帧中的值替换为聚合值，如mean
Code:
from pyspark import SparkConf, SparkContext
from pyspark.sql.functions import *
from pyspark.sql import Row, HiveContext, SQ
浏览 3提问于2017-09-28得票数 0
在“火花:最终指南”中，它说：
如果需要引用特定DataFrame的列，则可以在特定的DataFrame上使用col方法。
例如(在Python/Pyspark中)：
df.col("count")
但是，当我在包含列count的dataframe上运行后的代码时，就会得到错误'DataFrame' object has no attribute 'col'。如果我尝试column，我会得到一个类似的错误。
这本书是错的，还是我该怎么做呢？
我上的是星火2.3.1。dataframe是用以下内容创建的：
df = spark.read.f
浏览 2提问于2018-08-12得票数 9
下面是创建pyspark.sql DataFrame的代码
import numpy as np
import pandas as pd
from pyspark import SparkContext
from pyspark.sql import SQLContext
df = pd.DataFrame(np.array([[1,2,3],[4,5,6],[7,8,9],[10,11,12]]), columns=['a','b','c'])
sparkdf = sqlContext.createDataFrame(df, samplingR
浏览 0提问于2015-08-11得票数 6
2回答
我有一个小的PySpark DataFrame df
index col1
0 1
1 3
2 4
还有一本字典：
LOOKUP = {0: 2, 1: 5, 2: 5, 3: 4, 4: 6}
现在，我想在df中添加一个额外的列df，等于col1的LOOKUP值。
我的输出应该如下所示：
index col1 col2
0 1 5
1 3 4
2 4 6
我试着用：
df = df.withColumn(col("col2"), LOOKUP[co
浏览 7提问于2021-12-23得票数 3
回答已采纳
1回答
我正在编写一个在palantir中使用pyspark的代码，我有这个错误，我无法弄清楚。
错误是：
A TransformInput object does not have an attribute withColumn.
Please check the spelling and/or the datatype of the object.
我的代码供您参考
import pyspark.sql.functions as F
import pyspark.sql.types as T
from pyspark.sql.functions import when
from transfor
浏览 5提问于2022-09-17得票数 2
回答已采纳
我有一个用pyspark写的代码。我需要将其转换为字符串，然后将其转换为日期类型，等等。
我找不到任何方法来将此类型转换为字符串。我尝试了str()和.to_string()，但都不起作用。我把代码放在下面。
from pyspark.sql import functions as F
df = in_df.select('COL1')
> type(df)
> <class 'pyspark.sql.dataframe.DataFrame'>
> df.printSchema()
> |-- COL1: offset
浏览 7提问于2019-07-07得票数 0
回答已采纳
2回答
我已经搜索了一段时间了，如果有任何方法在Pyspark中使用一个Pyspark类，我还没有找到任何关于这个主题的文档或指南。
假设我在Scala中创建了一个简单的类，它使用了一些apache-spark库，如下所示：
class SimpleClass(sqlContext: SQLContext, df: DataFrame, column: String) {
def exe(): DataFrame = {
import sqlContext.implicits._
df.select(col(column))
}
}
有没有可能在Pyspark中使用这个类
浏览 3提问于2016-03-15得票数 29
回答已采纳
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
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
