pyspark 错误原因：java.lang.IllegalStateException：输入行没有架构所需的预期值数量答案

【问题标题】：pyspark Error Caused by: java.lang.IllegalStateException: Input row doesn't have expected number of values required by the schemapyspark 错误原因：java.lang.IllegalStateException：输入行没有架构所需的预期值数量
【发布时间】：2017-09-10 20:15:43
【问题描述】：

我有下面的 pyspark 代码来连接两个数据框。一切看起来都很简单，但是这个错误并没有输出。无法继续进行下去，您能否在此处帮助确定这个基本问题？

输入

C.csv

100,2015-09-03,SG,7
200,2016-01-30,AT,9
300,2016-01-25,AU,8
400,2016-01-22,AU,7

U.csv

248,248,COUNTRY,SG,Singapore
66,66,COUNTRY,AT,Austria
65,65,COUNTRY,AU,Australia

输出

100,Singapore
200,Austria
300,Australia
400,Australia

来源

pyspark 代码是：test.py

from pyspark import SparkConf, SparkContext
from pyspark.sql.types import StringType
from pyspark import SQLContext

conf = SparkConf().setAppName("HYBRID - READ CSV to HIVE ")
sc = SparkContext(conf=conf)

sqlContext = SQLContext(sc)
C_rdd = sc.textFile("./hybrid/C.csv").map(lambda line: line.split(","))
R_rdd = sc.textFile("./hybrid/U.csv").map(lambda line: line.encode("ascii", "ignore").split(","))

C_df = C_rdd.toDF(['C_No','Op_Dt','Try_Cd','Lb'])
R_df = R_rdd.toDF(['C_Id','P_Id','CC_Cd','C_Nm','C_Ds'])

New = C_df.join(R_df, C_df.Try_Cd == R_df.C_Nm).select(['C_No','C_Ds'])
New.show()

结果

Pyspark Error: $spark-submit  test.py
java.lang.IllegalStateException: Input row doesn't have expected number of values required by the schema. 5 fields are required while 6 values are provided.
        at org.apache.spark.sql.execution.EvaluatePython$.fromJava(python.scala:225)
        at org.apache.spark.sql.SQLContext$$anonfun$11.apply(SQLContext.scala:933)
        at org.apache.spark.sql.SQLContext$$anonfun$11.apply(SQLContext.scala:933)
        at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
        at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
        at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
        at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:149)

你能帮忙解决一下这里的问题吗？

【问题讨论】：

在执行R_rdd.toDF 语句时似乎抛出了错误。行 "248,248,COUNTRY,SG,Singapore 66,66,COUNTRY,AT,Austria 65,65,COUNTRY,AU,Australia 100,Singapore 200,Austria 300,Australia 400,Australia` ，它是连续的吗？拆分时会导致超过 5 列。
实际上我将所有这些行保留在不同的行中，但它们出现在此屏幕中。文件 C.csv 有 4 列，文件 U.csv 有 5 列数据。我期望的输出有两列，一列来自 file-1，另一列来自 file-2，通过连接这两个文件数据。但是由于某种原因它不起作用..如果有任何示例工作代码，您能否发送此案例？
这个解决方案解决了您的问题吗？

标签： python join dataframe pyspark

【解决方案1】：

希望你使用的是 spark 2.x + 然后试试这个 -

from pyspark.sql.types import StructType,StringType,IntegerType,StructField
from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("HYBRID - READ CSV to HIVE ") \
    .getOrCreate()

cSchema = StructType([StructField("C_No", IntegerType()),
                     StructField("Op_Dt", StringType()),
                     StructField("Try_Cd", StringType()),
                     StructField("Lb", IntegerType())])

uSchema = StructType([StructField("C_Id", IntegerType()),
                     StructField("P_Id", IntegerType()),
                     StructField("CC_Cd", StringType()),
                     StructField("C_Nm", StringType()),
                     StructField("C_Ds", StringType())])

c_df  = spark.read.csv("c.csv",schema=cSchema)
u_df  = spark.read.csv("u.csv",schema=uSchema)

New = c_df.join(u_df, c_df.Try_Cd == u_df.C_Nm).select(c_df.C_No,u_df.C_Ds)
New.show()

【讨论】：