python之Apache Spark (PySpark) 在读取 CSV 时处理空值

我正在尝试读取交通部的航类数据。它存储在 CSV 中，并不断获取 java.lang.NumberFormatException: null

我尝试将 nanValue 设置为空字符串，因为它的默认值为 NaN，但这没有用。

我当前的代码是:

spark = SparkSession.builder \ 
    .master('local') \ 
    .appName('Flight Delay') \ 
    .getOrCreate() 
 
schema = StructType([ 
    StructField('Year', IntegerType(), nullable=True), 
    StructField('Month', IntegerType(), nullable=True), 
    StructField('Day', IntegerType(), nullable=True), 
    StructField('Dow', IntegerType(), nullable=True), 
    StructField('CarrierId', StringType(), nullable=True), 
    StructField('Carrier', StringType(), nullable=True), 
    StructField('TailNum', StringType(), nullable=True), 
    StructField('Origin', StringType(), nullable=True), 
    StructField('Dest', StringType(), nullable=True), 
    StructField('CRSDepTime', IntegerType(), nullable=True), 
    StructField('DepTime', IntegerType(), nullable=True), 
    StructField('DepDelay', DoubleType(), nullable=True), 
    StructField('TaxiOut', DoubleType(), nullable=True), 
    StructField('TaxiIn', DoubleType(), nullable=True), 
    StructField('CRSArrTime', IntegerType(), nullable=True), 
    StructField('ArrTime', IntegerType(), nullable=True), 
    StructField('ArrDelay', DoubleType(), nullable=True), 
    StructField('Cancelled', DoubleType(), nullable=True), 
    StructField('CancellationCode', StringType(), nullable=True), 
    StructField('Diverted', DoubleType(), nullable=True), 
    StructField('CRSElapsedTime', DoubleType(), nullable=True), 
    StructField('ActualElapsedTime', DoubleType(), nullable=True), 
    StructField('AirTime', DoubleType(), nullable=True), 
    StructField('Distance', DoubleType(), nullable=True), 
    StructField('CarrierDelay', DoubleType(), nullable=True), 
    StructField('WeatherDelay', DoubleType(), nullable=True), 
    StructField('NASDelay', DoubleType(), nullable=True), 
    StructField('SecurityDelay', DoubleType(), nullable=True), 
    StructField('LateAircraftDelay', DoubleType(), nullable=True) 
]) 
 
flts = spark.read \ 
    .format('com.databricks.spark.csv') \ 
    .csv('/home/william/Projects/flight-delay/data/201601.csv', 
         schema=schema, nanValue='', header='true') 

这是我正在使用的 CSV:http://pastebin.com/waahrgqB

最后一行是中断并引发 java.lang.NumberFormatException: null

似乎有些数字列是空字符串，而有些只是空的。有人可以帮我解决这个问题吗？

请您参考如下方法:

感谢 KiranM 的建议，我得以找到解决方案。我让 Spark 推断架构(所有内容都设置为字符串)，然后手动设置我想要的列为数字。

代码如下:

from pyspark.sql import (SQLContext, 
                     SparkSession) 
 
from pyspark.sql.types import (StructType, 
                           StructField, 
                           DoubleType, 
                           IntegerType, 
                           StringType) 
 
spark = SparkSession.builder \ 
    .master('local') \ 
    .appName('Flight Delay') \ 
    .getOrCreate() 
 
 
flts = spark.read \ 
    .format('com.databricks.spark.csv') \ 
    .csv('/home/william/Projects/flight-delay/data/merged/2016.csv', 
         inferSchema='true', nanValue="", header='true', mode='PERMISSIVE') 
 
 
flts = flts \ 
    .withColumn('Year', flts['Year'].cast('int')) \ 
    .withColumn('Month', flts['Month'].cast('int')) \ 
    .withColumn('Day', flts['Day'].cast('int')) \ 
    .withColumn('Dow', flts['Dow'].cast('int')) \ 
    .withColumn('CRSDepTime', flts['CRSDepTime'].cast('int')) \ 
    .withColumn('DepTime', flts['DepTime'].cast('int')) \ 
    .withColumn('DepDelay', flts['DepDelay'].cast('int')) \ 
    .withColumn('TaxiOut', flts['TaxiOut'].cast('int')) \ 
    .withColumn('TaxiIn', flts['TaxiIn'].cast('int')) \ 
    .withColumn('CRSArrTime', flts['CRSArrTime'].cast('int')) \ 
    .withColumn('ArrTime', flts['ArrTime'].cast('int')) \ 
    .withColumn('ArrDelay', flts['ArrDelay'].cast('int')) \ 
    .withColumn('Cancelled', flts['Cancelled'].cast('int')) \ 
    .withColumn('Diverted', flts['Diverted'].cast('int')) \ 
    .withColumn('CRSElapsedTime', flts['CRSElapsedTime'].cast('int')) \ 
    .withColumn('ActualElapsedTime', flts['ActualElapsedTime'].cast('int')) \ 
    .withColumn('AirTime', flts['AirTime'].cast('int')) \ 
    .withColumn('Distance', flts['Distance'].cast('int')) \ 
    .withColumn('CarrierDelay', flts['CarrierDelay'].cast('int')) \ 
    .withColumn('WeatherDelay', flts['WeatherDelay'].cast('int')) \ 
    .withColumn('NASDelay', flts['NASDelay'].cast('int')) \ 
    .withColumn('SecurityDelay', flts['SecurityDelay'].cast('int')) \ 
    .withColumn('LateAircraftDelay ', flts['LateAircraftDelay '].cast('int')) 

也许我可以把它放到一个循环中，但我现在要用它来运行。