Pyspark，错误：输入没有模式所需的预期值数量和列后的额外尾随逗号

Posted 2023-04-15

【中文标题】Pyspark，错误：输入没有模式所需的预期值数量和列后的额外尾随逗号

【英文标题】：Pyspark, error:input doesn't have expected number of values required by the schema and extra trailing comma after columns

【发布时间】：2018-04-16 04:05:40

【问题描述】：

首先我制作了两个表（RDD）来使用以下命令

rdd1=sc.textFile('checkouts').map(lambda line:line.split(',')).map(lambda fields:((fields[0],fields[3],fields[5]), 1) )
rdd2=sc.textFile('inventory2').map(lambda line:line.split(',')).map(lambda fields:((fields[0],fields[8],fields[10]), 1) )

第一个 RDD 中的键是 BibNum、ItemCollection 和 CheckoutDateTime。当我检查第一个 RDD 的值以使用 rdd1.take(2) 时，它显示

[((u'BibNum', u'ItemCollection', u'CheckoutDateTime'), 1), ((u'1842225', u'namys', u'05/23/2005 03:20:00 PM'), 1)]

类似地，第二个 RDD 中的键是 BibNum、ItemCollection 和 Itemlocation。值如下：

[((u'BibNum', u'ItemCollection', u'ItemLocation'), 1), ((u'3011076', u'ncrdr', u'qna'), 1)]

创建两个 RDD 后，我尝试加入这两个以使用 rdd3=rdd1.join(rdd2) 之后，当我检查 rdd3 的值以使用 rdd3.take(2) 时。发生以下错误。

IndexError: list index out of range

我不知道为什么会这样。知道原因的请赐教。如果您对我的问题或代码有任何疑问，请告诉我，我会尽力澄清它。谢谢

编辑--- 我为每个 RDD 提供了我的示例输入数据

BibNum,ItemBarcode,ItemType,ItemCollection,CallNumber,CheckoutDateTime,,,,,,,
1842225,10035249209,acbk,namys,MYSTERY ELKINS1999,05/23/2005 03:20:00 PM,,,,,,,
1928264,10037335444,jcbk,ncpic,E TABACK,12/14/2005 05:56:00 PM,,,,,,,
1982511,10039952527,jcvhs,ncvidnf,VHS J796.2 KNOW_YO 2000,08/11/2005 01:52:00 PM,,,,,,,
2026467,10040985615,accd,nacd,CD 782.421642 Y71T,10/19/2005 07:47:00 PM,,,,,,,
2174698,10047696215,jcbk,ncpic,E KROSOCZ,12/29/2005 03:42:00 PM,,,,,,,
1602768,10028318730,jcbk,ncpic,E BLACK,10/08/2005 02:15:00 PM,,,,,,,
2285195,10053424767,accd,cacd,CD 782.42166 F19R,09/30/2005 10:16:00 AM,,,,Input,BinNumber,Date,BinNumber+Month
2245955,10048392665,jcbk,ncnf,J949.73 Or77S 2004,12/05/2005 05:03:00 PM,,,,,,,
770918,10044828100,jcbk,ncpic,E HILL,07/22/2005 03:17:00 PM,,,,,,,

.

BibNum,Title,Author,ISBN,PublicationYear,Publisher,Subjects,ItemType,ItemCollection,FloatingItem,ItemLocation,ReportDate,ItemCount,,,,,,,,,,,,,
3011076,A tale of two friends / adapted by Ellie O'Ryan ; illustrated by Tom Caulfield| Frederick Gardner| Megan Petasky| and Allen Tam.,O'Ryan| Ellie,1481425730| 1481425749| 9781481425735| 9781481425742,2014,Simon Spotlight|,Musicians Fiction| Bullfighters Fiction| Best friends Fiction| Friendship Fiction| Adventure and adventurers Fiction,jcbk,ncrdr,Floating,qna,09/01/2017,1,,,,,,,,,,,,,
2248846,Naruto. Vol. 1| Uzumaki Naruto / story and art by Masashi Kishimoto ; [English adaptation by Jo Duffy].,Kishimoto| Masashi| 1974-,1569319006,2003| c1999.,Viz|,Ninja Japan Comic books strips etc| Comic books strips etc Japan Translations into English| Graphic novels,acbk,nycomic,NA,lcy,09/01/2017,1,,,,,,,,,,,,,
3209270,Peace| love & Wi-Fi : a ZITS treasury / by Jerry Scott and Jim Borgman.,Scott| Jerry| 1955-,144945867X| 9781449458676,2014,Andrews McMeel Publishing|,Duncan Jeremy Fictitious character Comic books strips etc| Teenagers United States Comic books strips etc| Parent and teenager Comic books strips etc| Families Comic books strips etc| Comic books strips etc| Comics Graphic works| Humorous comics,acbk,nycomic,NA,bea,09/01/2017,1,,,,,,,,,,,,,
1907265,The Paris pilgrims : a novel / Clancy Carlile.,Carlile| Clancy| 1930-,786706155,c1999.,Carroll & Graf|,Hemingway Ernest 1899 1961 Fiction| Biographical fiction| Historical fiction,acbk,cafic,NA,cen,09/01/2017,1,,,,,,,,,,,,,
1644616,Erotic by nature : a celebration of life| of love| and of our wonderful bodies / edited by David Steinberg.,,094020813X,1991| c1988.,Red Alder Books/Down There Press|,Erotic literature American| American literature 20th century,acbk,canf,NA,cen,09/01/2017,1,,,,,,,,,,,,,

编辑---------------------------------------------- date_count --> DataFrame[BibNum : string, ItemCollection : string, CheckoutDateTime : string, count : BigInt ] .. 显示喜欢这个 但是当我使用 date_count.take(2) 检查它的值时，它显示如下错误：输入没有架构所需的预期值数量。需要 6 个字段，提供 7 个值。

df_final 架构如下所示：DataFrame[BibNum:string, ItemType:string, ItemCollection:string, ItemBarcode:string, CallNumber:string, CheckoutDateTime:string, Title:string, Author:string, ISBN:string, PublicationYear:string, Publisher:string, Subjects:string, FloatingItem:string, ItemLocation:string, ReportDate:string, ItemLocation:string, : string , : string, : string .... : string , : string]

【问题讨论】：

请把数据打印出来给我们看。

你的意思是输入数据？..

我把我的输入数据放到了写作中:)

@pissall 我不确定你指的是哪个数据，但如果你指的是每个 RDD 的数据，我已经在诸如 [((u'BibNum', u'ItemCollection' , u'CheckoutDateTime'), 1), ((u'1842225', u'namys', u'05/23/2005 03:20:00 PM'), 1)]

我明白了。我不明白的是你为什么要这样映射它？我的意思是使用 map(lambda fields:((fields[0],fields[3],fields[5]), 1))

【参考方案1】：

所以我会试着回答你的问题。解决方案可能在语法上很混乱，但我会尽力做到最好（我现在没有环境可以测试）。如果这是您正在寻找的，请告诉我，否则我可以帮助您微调解决方案。

这是Join in Pyspark的文档

所以当你阅读文件时：

rdd1=sc.textFile('checkouts').map(lambda line:line.split(','))
rdd2=sc.textFile('inventory2').map(lambda line:line.split(','))
# Define the headers for both the files
rdd1_header = rdd1.first()
rdd2_header = rdd2.first()

# Define the dataframe
rdd1_df = rdd1.filter(lambda line: line != rdd1_header).toDF(rdd1_header)
rdd2_df = rdd2.filter(lambda line: line != rdd2_header).toDF(rdd2_header)

common_cols = [x for x in rdd1_df.columns if x in rdd2_df.columns]

df_final = rdd1_df.join(rdd2_df, on=common_cols)
date_count = df_final.groupBy(["BibNum", "ItemCollection", "CheckoutDateTime"]).count()

编辑：

1) 您的错误："pyspark.sql.utils.AnalysisException: u"Reference 'ItemCollection' is ambiguous, could be ItemCollection#3, ItemCollection#21" 是由于连接后生成了多个列。你需要什么要做的是将所有常用列包含在您的联接中。我将在代码中提及。

2) 另一个问题：在每个RDD的最后部分添加了一些奇怪的部分，例如-- [Row(BibNum=u'1842225', ItemBarcode=u'10035249209', ItemType=u'acbk', ItemCollection=u'namys', CallNumber=u'MYSTERY ELKINS1999', CheckoutDateTime=u'05/23/2005 03:20:00 PM', =u'', =u'', =u'', =u' ', =u'', =u'', =u'')

为此，您提到您的 CSV 文件如下：

BibNum,ItemBarcode,ItemType,ItemCollection,CallNumber,CheckoutDateTime,,,,,,,
1842225,10035249209,acbk,namys,MYSTERY ELKINS1999,05/23/2005 03:20:00 PM,,,,,,,
1928264,10037335444,jcbk,ncpic,E TABACK,12/14/2005 05:56:00 PM,,,,,,,

现在，如果您可以看到日期列后面有很多尾随逗号。即',,,,,,,,'，它给出了那些额外的空列（在逗号拆分之后），您可以删除它们。

【讨论】：

先谢谢你的回答，但你知道这个错误是什么吗？-- join() 得到了一个意外的关键字参数 'on'

可能你正在使用其他的 join() 方法。我已经添加了文档链接。它提到了“on”和“how”的使用，以便您提及您希望如何执行连接。您可以使用其他方式加入，以 (rdd1_df.BibNum == rdd2_df.BibNum) 为条件。

先生，实际上我使用了cloudera，但它不适用于'on'命令，所以我在谷歌云平台上尝试了它，所以效果很好。但其他问题发生如下：pyspark.sql.utils.AnalysisException: u"Reference 'ItemCollection' is ambiguous, could be ItemCollection#3, ItemCollection#21.;"

在每个RDD的最后部分添加了一些奇怪的部分，比如--[Row(BibNum=u'1842225', ItemBarcode=u'10035249209', ItemType=u'acbk', ItemCollection=u'namys', CallNumber=u'MYSTERY ELKINS1999', CheckoutDateTime=u'05/23/2005 03:20:00 PM', =u'', =u'', =u'', =u' ', =u'', =u'', =u''),

在传递给 'on' 的列表中包含所有常见的列，这样它就不会产生重复的列。它正在制作重复的项目集合列