如何处理spark sql中缺失的列

本文介绍了如何处理spark sql中缺失的列的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！

问题描述

限时送ChatGPT账号..

我们正在处理无模式 JSON 数据，有时 Spark 作业会失败，因为我们在 Spark SQL 中引用的某些列在一天中的某些小时不可用.在这些时间里，spark 作业失败，因为被引用的列在数据框中不可用.如何处理这种情况?我尝试过 UDF，但我们缺少太多列，因此无法真正检查每一列的可用性.我还尝试在更大的数据集上推断模式并将其应用于数据框，期望缺失的列将填充为空，但模式应用程序失败并出现奇怪的错误.

We are dealing with schema free JSON data and sometimes the spark jobs are failing as some of the columns we refer in spark SQL are not available for certain hours in the day. During these hours the spark job fails as the column being referred is not available in the data frame. How to handle this scenario? I have tried UDF but we have too many columns missing so can't really check each and every column for availability. I have also tried inferring a schema on a larger data set and applied it on the data frame expecting that missing columns will be filled with null but the schema application fails with weird errors.

请推荐

推荐答案

这对我有用.创建了一个函数来检查所有预期的列并将列添加到数据框中(如果缺失)

This worked for me. Created a function to check all expected columns and add columns to dataframe if it is missing

def checkAvailableColumns(df: DataFrame, expectedColumnsInput: List[String]) : DataFrame = {
    expectedColumnsInput.foldLeft(df) {
        (df,column) => {
            if(df.columns.contains(column) == false) {
                df.withColumn(column,lit(null).cast(StringType))
            }
            else (df)
        }
    }
}

val expectedColumns = List("newcol1","newcol2","newcol3")

val finalDf = checkAvailableColumns(castedDateSessions,expectedColumns)

这篇关于如何处理spark sql中缺失的列的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！