0

I need to validate the each column of a data frame with expected length or not. If not, one new column need to be added to the data frame which populates consolidation the validation results. I need this to be written using python.

For example, Below is my data frame structure. And my validation criteria is total length of any columns should not exceed more than 3.

[col1   col2    col3
=========================
AAA     BBB     CCC
DDDD    EEE     BBBB
AAA     EEEE    BBBB

And I am expecting output as below.

col1    col2    col3                   length_check
======================================================================================
AAA     BBB     CCC      
DDDD    EEE     BBBB     Expected Length of col1 is 3; Expected Length of col3 is 3
AAA     EEEE    BBBB     Expected Length of col2 is 3; Expected Length of col3 is 3

Your inputs are much appreciated. Thanks

Code:

valid_rdd=parsed_file.map(lambda line: line if len(line)==4 else False)
                     .filter(lambda line:line!=False)
invalid_rdd=parsed_file.map(lambda line: line if len(line)!=4 else False)
                       .filter(lambda line:line!=False) 
columns=['colA','colB','colC','colD'] 
df_valid=valid_rdd.toDF(columns) 
df1=df_valid.withColumn('length_check', (f.when (f.length('colA')== 1, "True").otherwise("Expected Column length 1 but found "+ str(f.length('colA') + str(df_valid.colA) ))))
0

1 Answer 1

0

Well first of all, I'm not very experienced with PySpark also have little experience with Python, but I was able to replicate your code (with a little bit of refactoring).

Before I show you the code, I think it's important to mention that the code below is based on this article trying to replicate a real world scenario (I assume you'll manage to get all the necessary data to make it work).

# Assuming this is the returned value from the Data Frame (e.g.parsed_file.columns)
DATA_FRAME_COLUMNS = [
    'Col1',
    'Col2',
    'Col3',
]


def get_rows_from_a_column(column):
     # This dictionary is actually a mock of what should be something like parsed_file.select(column).rows
     return {
         'Col1': ['AAA', 'BBB', 'CCC'],
         'Col2': ['DDDD', 'EEE', 'BBBB'],
         'Col3': ['AAA', 'EEEE', 'BBBB']
     }[column]


def check_column_length(data_frame_columns):
    """
    This function checks for the length of rows that you have accordingly to 
    DATA_FRAME_COLUMNS
    """
    data = {}

    for column in data_frame_columns:
        column_rows = get_rows_from_a_column(column)

        if len(column_rows) != 3:
            raise ValueError('Expected length of ' + column + ' to be 3')

        data[column] = column_rows

    return data


print(check_column_length(DATA_FRAME_COLUMNS)) # {'Col2': ['DDDD', 'EEE', 'BBBB'], 'Col3': ['AAA', 'EEEE', 'BBBB'], 'Col1': ['AAA', 'BBB', 'CCC']}

If my advice is worth something, I would recommend you to have a look to the Zen of Python, I don't know your level with the language (but I'll assume you've heard about it anyway).

Why am I saying this? I think lambda functions are good in some cases but they usually go against the Explicit is better than implicit. principle.

This stuff is relative to your overall code by the way, how it looks and how easy it is to understand. I personally would avoid using it (but that depends on the situation).

PS: I wasn't sure until @barbsan's update on this post, if you wanted to check the length of rows of the column or the length of characters in a row. So I assume you want the first option.

Let me know if I missed anything.

1
  • Hi Joao, Thanks for your help. The suggested approach will not help because you are finding the length of each column array. But I am looking for an approach where length of value in each column should be checked and corresponding error message to be populated. Also I need error column to be added and number of columns in the data frame shouldn't be fixed to certain list.
    – Prabu K
    Commented Feb 25, 2019 at 23:23

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.