Pyspark使用列作为lookup - TypeError：‘groupBy’对象不可调用
Pyspark是一个用于大规模数据处理的Python库，它提供了丰富的功能和工具来处理和分析大数据集。在Pyspark中，使用列作为lookup是一种常见的操作，用于根据某一列的值来查找和聚合数据。
在你提供的问题中，出现了一个错误提示：TypeError: 'groupBy'对象不可调用。这个错误通常是因为在使用groupBy函数时，没有正确地使用聚合函数来对分组后的数据进行操作。
解决这个问题的方法是，在groupBy函数之后使用聚合函数，例如sum、count、avg等来对分组后的数据进行聚合操作。下面是一个示例代码：
from pyspark.sql import SparkSession
from pyspark.sql.functions import sum
# 创建SparkSession
spark = SparkSession.builder.getOrCreate()
# 读取数据
data = spark.read.csv("data.csv", header=True, inferSchema=True)
# 使用groupBy和sum函数对数据进行分组和聚合
result = data.groupBy("column_name").agg(sum("value"))
# 显示结果
result.show()在上面的示例代码中，我们首先创建了一个SparkSession对象，然后使用read.csv函数读取数据。接下来，使用groupBy函数对指定的列进行分组，然后使用agg函数和sum函数对分组后的数据进行求和操作。最后，使用show函数显示结果。
对于这个问题，Pyspark提供了一些相关的函数和方法来处理和操作数据，例如groupBy、agg、sum等。你可以根据具体的需求选择合适的函数和方法来完成相应的操作。
关于Pyspark的更多信息和详细介绍，你可以参考腾讯云的相关产品和文档：
希望以上信息能够帮助到你，如果还有其他问题，请随时提问。
相关·内容
1回答
我怎样才能做到这一点？
from pyspark.sql import functions as F
from pyspark.context import SparkContext
from pyspark.sql.session import SparkSession
sc = SparkContext('local')
spark = SparkSession(sc)
grouped=df.groupby([col list]).agg(F.count([col list]))
我读过关于堆栈溢出的类似问题，但没有找到确切的答案。
即使我试着把一列
grouped=dfn
浏览 3提问于2020-02-26得票数 0
回答已采纳
我有PySpark dataframe (source_df)，其中有一个值为逗号分隔的列。我正在尝试使用基于另一个数据访问(lookup_df)的查找来替换这些值。
source_df
A B T ... followed by N unrelated columns...
foo a,b,c sam
bar k,a,c bob
faz b,a,f sam
lookup_df
C D
a h1
b h2
c h3
输出数据
A T B new_col ... followed by N unrelated colu
浏览 5提问于2021-02-10得票数 3
回答已采纳
1回答
我想除以pyspark中两列的和。例如，我有如下所示的数据集：
A B C
1 1 2 3
2 1 2 3
3 1 2 3
我希望得到colA之和除以colB之和，如下所示：
6 (Sum of colB) / 3 (Sum of colA) = 2
我试过这样做：
sumofA = df.groupby().sum('A')
sumofB = df.groupby().sum('B')
Result = B / A
但是它会产生这样的错误：
TypeError: unsupported operand type(s
浏览 2提问于2020-03-27得票数 3
回答已采纳
我正在查找数据中所有列的“模式”，并将它们存储在列表中。计算每个列的我的模式的代码：
from pyspark.sql.functions import *
#calculating mode value
mode_val = []
for i in df_num.columns :
cnts = df_num.groupBy(i).count()
mode = cnts.join(
cnts.agg(max("count").alias("max_")), col("count") == col(&#
浏览 0提问于2018-04-03得票数 2
回答已采纳
我的代码是：
import pyspark
from pyspark.sql import SparkSession
from pyspark.conf import SparkConf
from pyspark import SparkContext
spark = SparkSession.builder \
.master("local") \
.appName("pyspark_uygulama") \
.getOrCreate()
sc = spark.sparkContext()
sc
我得到了一个错误：
--------
浏览 1提问于2020-10-20得票数 0
回答已采纳
我无法对数据帧使用筛选器。我一直收到错误“TypeError(”条件应该是字符串或列“)” 我已经尝试更改滤镜以使用col对象。尽管如此，它还是不起作用。 path = 'dbfs:/FileStore/tables/TravelData.txt'
data = spark.read.text(path)
from pyspark.sql.types import StructType, StructField, IntegerType , StringType, DoubleType
schema = StructType([
StructField("from
浏览 21提问于2019-02-02得票数 0
回答已采纳
我在pyspark中有一个列，其中包含如下所示的映射数组：
[{"address": "Fadden", "city": "", "country": "", "note": "", "stateProvince": "Queensland"}]
df.printSchema()为该列返回以下内容：
|-- constituencies: array (nullable = true)
| |-- element: struc
浏览 1提问于2022-02-20得票数 3
回答已采纳
2回答
我有一个具有如下标题的数据集：
|State|2020-01-22|2020-01-23|2020-01-24|2020-01-25|2020-01-26|2020-01-27|2020-01-28|
我正在尝试基于groupBy列和每一列的行值之和(列数保持不变)进行groupBy。但当我用：
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
df = df.groupBy('State').agg(F.sum())
但是我得到了错误：sum() missing 1 required
浏览 3提问于2022-04-19得票数 0
3回答
当我尝试groupBy并获取最大值时，有了这个数据帧I正在获取列是不可迭代的：
linesWithSparkDF
+---+-----+
| id|cycle|
+---+-----+
| 31| 26|
| 31| 28|
| 31| 29|
| 31| 97|
| 31| 98|
| 31| 100|
| 31| 101|
| 31| 111|
| 31| 112|
| 31| 113|
+---+-----+
only showing top 10 rows
ipython-input-41-373452512490> in runlgmodel2
浏览 4提问于2016-04-29得票数 23
回答已采纳
我正在将一个pySpark实现迁移到Pandas。为了转换大型数据集，我使用了dask包。 pySpark实现： df.groupBy('Key').agg(collect_set('ColumnA').alias('Items'), collect_set('ColumnB').alias('DocumentId')) 到Pandas df.groupby("Key")[['ColumnA','ColumnB']].agg(set).reset_index() 然
浏览 31提问于2021-08-16得票数 0
1回答
我有一个python类，它的函数如下：
class Features():
def __init__(self, json):
self.json = json
def get_email(self):
email = self.json.get('fields', {}).get('email', None)
return email
我正尝试在一个pyspark数据帧中使用get_email函数，基于另一个列"raw_json“创建一个新列，该列由json值组成：
df = data.
浏览 6提问于2020-11-12得票数 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
回答已采纳
1回答
熊猫群在一列名单上
、、、
我有一个包含pandas的列的lists数据
df = pd.DataFrame({'List': [['once', 'upon'], ['once', 'upon'], ['a', 'time'], ['there', 'was'], ['a', 'time']], 'Count': [2, 3, 4, 1, 2]})
Count List
2 [once, upon]
3 [onc
浏览 3提问于2018-03-22得票数 7
回答已采纳
我使用itertools.groupby编写了一个映射函数来聚合数据，我所做的如下所示。
驱动程序代码
pair_count = df.mapPartitions(lambda iterable: pair_func_cnt(iterable))
pair_count.collection()
映射函数
def pair_func_cnt(iterable):
from itertools import groupby
ls = [[1,2,3],[1,2,5],[1,3,5],[2,4,6]]
grp1 = [(k,g) for k,g in groupby(ls,
浏览 2提问于2016-08-05得票数 0
2回答
简写版：
如何预览火花放电外壳中的列？
我有一个pyspark.sql.column.Column类型的对象pyspark.sql.column.Column，当我执行a.show()时，我得到了TypeError: 'Column' object is not callable。
想知道是否有类似于熊猫功能的火花放电，这样我就可以在将列转换为pyspark之后使用show()。
长版本：
我有以下数据集为df
+----------+-------------------+--------+
| number| p_efc_dtm|cus_type|
浏览 0提问于2018-03-09得票数 0
回答已采纳
我想向一个包含来自另一个数据帧(df2)的聚合信息的pyspark数据帧(df1)添加一个新列。
df1.show()
+----------------+
| name |
+----------------+
| 000097 |
| 000097 |
| 000098 |
+----------------+
df2.show()
+----------------+----------------+
| name | id |
+----------------+---
浏览 6提问于2019-07-13得票数 0
1回答
我想测试一个列中的值是否存在于常规python中，或者是在一个when().otherwise()代码块中是否存在吡火花映射，但无法确定正确的语法。使用"Count“列的排列将有多个There ()子句，因此需要类似于”if/elif/elif“的内容。字典/地图将很大，而不是数据文件中的列。
from pyspark.sql import SparkSession, Row
import pyspark.sql.functions as F
from pyspark.sql.types import *
from itertools import chain
spark = (Sp
浏览 2提问于2022-04-21得票数 2
回答已采纳
1回答
我们有一个1.5BM记录分散在几个csv文件。为了生成一个count聚合，我们需要对几个列进行分组。
我们目前的战略是：
将它们加载到数据文件中(使用Dask或pyspark)
聚合列，以便生成2列作为键:value(我们不确定这是否值得)
将文件保存为Parquet
读取Parquet文件(Dask或pyspark)，并在dataframe的索引上运行groupby。
对于Parquet文件上高效的groupby来说，最佳实践是什么？
在索引上而不是在列(或一组列)上执行groupby有多大益处？
我们知道有一个partition可以提供帮助--但在我们的例子中，我们
浏览 2提问于2017-07-09得票数 3
4回答
我有这些列为id、price、timestamp的数据。
我希望找到按id分组的中值。
我正在使用这个代码来找到它，但是它给了我这个错误。
from pyspark.sql import DataFrameStatFunctions as statFunc
windowSpec = Window.partitionBy("id")
median = statFunc.approxQuantile("price",
[0.5],
0)
浏览 20提问于2017-07-24得票数 12
回答已采纳
在应用调用AWS API的Pyspark UDF时，我收到以下错误
PicklingError: Could not serialize object: TypeError: can't pickle SSLContext objects
代码是
import pyspark.sql.functions as sqlf
import boto3
comprehend = boto3.client('comprehend', region_name='us-east-1')
def detect_sentiment(text):
response
浏览 61提问于2020-05-24得票数 1
我对PySpark环境很陌生，在尝试用加密模块加密RDD中的数据时遇到了一个错误。下面是代码：
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('encrypt').getOrCreate()
df = spark.read.csv('test.csv', inferSchema = True, header = True)
df.show()
df.printSchema()
from cryptography.fernet import Fernet
ke
浏览 0提问于2017-08-21得票数 8
回答已采纳
我试图在dataframe中对数字列进行排序，但是得到这个错误的'id‘列在特定的站点上有id的计数。例如2272、2202、1855等
df.sort_values(by=['id'])
但是，我得到了这个错误：
‘无法访问'DataFrameGroupBy’对象的可调用属性‘DataFrameGroupBy’，尝试使用'apply‘方法
浏览 1提问于2018-08-09得票数 3
3回答
有两列：
ID Text
1 a
2 b
3 c
我怎样才能用下面这样的伪变量创建矩阵：
ID a b c
1 1 0 0
2 0 1 0
3 0 0 1
使用pyspark库及其特性？
浏览 0提问于2016-03-09得票数 13
回答已采纳
我必须在pyspark数据帧上进行2级分组。我的试探性的：
grouped_df=df.groupby(["A","B","C"])
grouped_df.groupby(["C"]).count()
但我得到以下错误：
'GroupedData' object has no attribute 'groupby'
我想我应该首先将分组的对象转换为pySpark DF。但我不能这么做。
有什么建议吗？
浏览 2提问于2017-10-18得票数 7
回答已采纳
所以我有一个给定的PySpark DataFrame，比如df，如下所示： df.show()
+--------------------+-------------------+
| series| value |
+--------------------+-------------------+
| XXXX-AAAA | 1 |
| XXXX-BB | 2 |
| XXXX-CCCCC | 3
浏览 101提问于2021-01-15得票数 0
回答已采纳
我想按pyspark数据帧分组，并计算特定列的方差。对于一般人来说，这很容易，可以这样做
from pyspark.sql import functions as func
AVERAGES=df.groupby('country').agg(func.avg('clicks').alias('avg_clicks')).collect()
然而，对于方差，函数子模块中似乎没有任何聚合函数(我也想知道为什么，因为这是一个相当常见的操作)。
浏览 7提问于2015-08-12得票数 5
2回答
要Py的SQL查询(Spark)
、、、、
我有以下SQL查询，我想将其转换为pyspark。我想使用两个列pp和gender，并在pyspark中执行以下操作 %sql
SELECT pp
, SUM(CASE WHEN Gender = 'M'
THEN 1.0 ELSE 0.0 END) /
COUNT(1) AS gender_score
, count(1) AS total
FROM df
WHERE gender in ('M', 'F')
GROUP BY pp
HAVING
浏览 12提问于2020-12-16得票数 0
我有一个字典的RDD，我想得到一个只包含不同元素的RDD。但是，当我试图打电话给
rdd.distinct()
PySpark给出了以下错误
TypeError: unhashable type: 'dict'
at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)
at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)
at org.ap
浏览 5提问于2016-02-19得票数 2
回答已采纳
1回答
我有一个列id，它的类型是int，但后来更改为bigint。它具有这两种类型的值。 from pyspark.sql.functions import *
from pyspark.sql.types import *
df = spark.read.parquet('hdfs path')
df = df.select("id", "code")
df=df.withColumn("id1", df["id"].cast(LongType()))
res1=df.select("id1",
浏览 107提问于2021-01-11得票数 0
我正在应用一个映射，然后使用pyspark在RDD上应用reduceByKey转换。我尝试了以下两种语法，这两种语法似乎都有效：
案例1：
my_rdd_out = my_rdd.map(lambda r: [r['my_id'], [[r['my_value']]]])\
.reduceByKey(lambda a, b: a+b)\
.map(lambda r: r[1])
案例2：
my_rdd_out = my_rdd.map(
浏览 0提问于2016-09-26得票数 0
回答已采纳
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
回答已采纳
1回答
解析电火花中的地址函数
、、、、
我想在PySpark中的dataframe中添加一个列，其中包含通过libpostal库解析的地址。
import pyspark.sql.functions as sf
from postal.parser import parse_address
df = spark.read.parquet(path_hdfs)
df = df.select("id", "name" ,"street", "cty")\
.withColumn("address", parse_address(sf.conca
浏览 1提问于2021-04-19得票数 2
我需要提取csv文件中指定列中的指定字符，我看到一个错误："TypeError: tuple索引必须是整数或切片，而不是str“。
我的csv文件
name,age,sex
li,26,M
mu,30,F
import os
import pandas as pd
import glob
for i in glob.glob('H:\\1\\*.csv'):
result_1=pd.read_csv(i,encoding="utf-8",dtype=object)
SubNetwork_list=list(result_1.groupby
浏览 6提问于2022-07-11得票数 0
我发现PySpark有一个名为drop的方法，但它似乎一次只能删除一列。关于如何同时删除多个列有什么想法吗？
df.drop(['col1','col2'])
TypeError Traceback (most recent call last)
<ipython-input-96-653b0465e457> in <module>()
----> 1 selectedMachineView = machineView.drop([['GpuName',
浏览 1提问于2016-02-28得票数 42
回答已采纳
2回答
在groupBy.agg()中创建字典是可能的吗？下面是一个玩具示例：
import pyspark
from pyspark.sql import Row
import pyspark.sql.functions as F
sc = pyspark.SparkContext()
spark = pyspark.sql.SparkSession(sc)
toy_data = spark.createDataFrame([
Row(id=1, key='a', value="123"),
Row(id=1, key='b', v
浏览 0提问于2019-03-22得票数 6
1回答
我希望在以前创建的列表中做一个减缩。在输出中，结果显示"'map‘对象不可调用“，或者删除它返回的列表"<map对象at 0x7fc398d98670>”
我不知道这个错误是从哪里来的。
import pyspark
from pyspark.sql import SparkSession, Row
from pyspark.sql.types import MapType, StringType
from pyspark.sql.functions import col
from pyspark.sql.types import StructTy
浏览 7提问于2022-07-19得票数 0
1回答
我试图将四列(QBR、码、触地和截取)连接或组合到一个列中，并将它们按球衣编号分组，将sql函数作为pyspark中的f。下面列出了我试图使用的编码、实际数据和预期的数据结果。
import pyspark.sql.functions as f
from pyspark.sql.functions import concat, lit, col
df = df.groupby('Jersey Number).withColumn("joined", f.concat(f.col('QBR'), f.lit(','), f.col('
浏览 5提问于2020-09-08得票数 0
回答已采纳
我正在尝试执行一个与本文档中的示例非常相似的groupby过滤器：
>>> df = pd.DataFrame({'A' : ['foo', 'bar', 'foo', 'bar',
... 'foo', 'bar'],
... 'B' : [1, 2, 3, 4, 5, 6],
... 'C' : [2.0
浏览 0提问于2019-02-15得票数 12
回答已采纳
我试图使用pyspark.pandas API在数据库上构造数据的共生矩阵。
我试着用这个方法构造矩阵。
该代码在熊猫中运行良好，但在使用pyspark.pandas时出现了错误。
coocc = psdf.T.dot(psdf)
coocc
我得到了这个错误
TypeError: Unsupported type DataFrame
我查过医生了。
pyspark.pandas.DataFrame.dot()
以串联作为输入。
我尝试使用psdf.squeeze()将dataframe转换为系列化，但它并不将dataframe转换为串联，因为我的dataframe有多个列。
有没有办法将py
浏览 8提问于2022-10-14得票数 0
回答已采纳
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
回答已采纳
因此，我使用Spark来进行情感分析，并且我一直在使用序列化器(我认为)来传递python对象时出现错误。
PySpark worker failed with exception:
Traceback (most recent call last):
File "/Users/abdul/Desktop/RSI/spark-1.0.1-bin- hadoop1/python/pyspark/worker.py", line 77, in main
serializer.dump_stream(func(split_index, iterator), outf
浏览 0提问于2014-07-25得票数 7
我正在尝试应用一个带有两个参数的pandas_udf。但是我得到了这个错误。首先，我尝试使用一个参数，它是可以的： from pyspark.sql.functions import pandas_udf, PandasUDFType
from pyspark.sql import SparkSession
from pyspark.sql.types import *
spark = SparkSession \
.builder \
.config('spark.cores.max', 100) \
.getOrCreat
浏览 24提问于2018-12-19得票数 3
4回答
我正在使用这个数据集
代码：
df = pd.read_csv('Data/census.csv')
df = df[df['SUMLEV']==50]
print(df.head())
我的数据是这样的：
SUMLEV REGION DIVISION ... RNETMIG2013 RNETMIG2014 RNETMIG2015
1 50 3 6 ... -2.722002 2.592270 -2.187333
2 50 3 6 ... 2
浏览 3提问于2020-07-19得票数 6
回答已采纳
我有如下所示的数据集：
我是按年龄分组的，平均每个年龄的朋友数。
from pyspark.sql import SparkSession
from pyspark.sql import Row
import pyspark.sql.functions as F
def parseInput(line):
fields = line.split(',')
return Row(age = int(fields[2]), numFriends = int(fields[3]))
spark = SparkSession.builder.appName
浏览 3提问于2020-07-11得票数 5
回答已采纳
2回答
如何生产熊猫数据行以激发数据
、、、、
嗨，我正在进行转换，我已经创建了some_function(iter)生成器到yield Row(id=index, api=row['api'], A=row['A'], B=row['B']，以生成从熊猫数据格式到rdd的转换行，并激发数据格式。我搞错了。(我必须使用熊猫来转换数据，因为有大量的遗留代码)
输入火花DataFrame
respond_sdf.show()
+-------------------------------------------------------------------+
|content
浏览 5提问于2020-12-22得票数 2
回答已采纳
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
回答已采纳
我有以下python/pandas命令：
df.groupby('Column_Name').agg(lambda x: x.value_counts().max()
我在这里获取DataFrameGroupBy对象中所有列的值计数。
如何在PySpark中执行此操作？
浏览 1提问于2018-06-27得票数 32
import pandas as pd
import numpy as np
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql.functions import PandasUDFType, pandas_udf
from pyspark.sql.types import *
import os
@pandas_udf(schema, functionType=PandasUDFType.GROUPED_MAP)
def split(df, validation_
浏览 5提问于2020-09-21得票数 0
我使用以下pyspark逻辑对一些目标列进行分组，然后将另一个目标列收集到一个数组中：
(
df
.groupBy(groupby_cols)
.agg(
F.collect_list(
F.col(target_col)
).alias(target_col)
)
)
我希望将结果限制为每个收集的列表最多保留N值，以便结果目标列由具有最大长度为N的数组的单元格组成。
现在，我可以在pyspark中用一个UDF来实现这一点，它接受target_col并在每个单元上应用一个lambda：lambda x: x
浏览 8提问于2020-10-09得票数 1
我正在尝试在Python中实现一个缓存函数。代码如下所示：
def memoize(func):
"""Store the results of the decorated function for fast lookup
"""
# Store results in a dict that maps arguments to results
cache = {}
def wrapper(*args, **kwargs):
# If these arguments haven't
浏览 2提问于2020-11-02得票数 3
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
