Comment gérer les exceptions dans Pyspark pour les problèmes de science des données ?

Question

Comment identifier le type d'exception que le renommage des colonnes va produire et comment le gérer dans pyspark :

def rename_columnsName(df, columns):   #provide names in dictionary format
if isinstance(columns, dict):     
    for old_name, new_name in columns.items():
        df = df.withColumnRenamed(old_name, new_name)
    return df.show()
else:
    raise ValueError("'columns' should be a dict, like {'old_name':'new_name', 'old_name_one more':'new_name_1'}")

comment le tester en générant une exception avec un ensemble de données.

Answer 1

Voici un exemple de test d'une fonction PySpark qui lève une exception. Dans cet exemple, nous vérifions qu'une exception est levée si l'ordre de tri est "cats" .

def it_throws_an_error_if_the_sort_order_is_invalid(spark):
    source_df = spark.create_df(
        [
            ("jose", "oak", "switch"),
            ("li", "redwood", "xbox"),
            ("luisa", "maple", "ps4"),
        ],
        [
            ("name", StringType(), True),
            ("tree", StringType(), True),
            ("gaming_system", StringType(), True),
        ]
    )
    with pytest.raises(ValueError) as excinfo:
        quinn.sort_columns(source_df, "cats")
    assert excinfo.value.args[0] == "['asc', 'desc'] are the only valid sort orders and you entered a sort order of 'cats'"

Notez que le test vérifie le message d'erreur spécifique qui est fourni.

Vous pouvez fournir des données non valides à votre rename_columnsName et vérifiez que le message d'erreur correspond à ce que vous attendez.

Quelques autres conseils :

• suivre les exemples pour renommer les colonnes aquí y aquí . Vous ne devez pas appeler withColumnRenamed dans une boucle.
• Écrire des transformations de DataFrame à l'aide de le format de transformation standard afin qu'ils puissent être enchaînés avec DataFrame#transform
• utiliser pytest-describe d'organiser ces types de tests
• Vérifier ce fichier test pour une série d'exemples

Answer 2

J'ai trouvé la solution à cette question, nous pouvons gérer les exceptions dans Pyspark de la même manière que dans Python. eg :

def rename_columnsName(df, columns):#provide names in dictionary format
try:

   if isinstance(columns, dict):
      for old_name, new_name in columns.items():     

           df = df.withColumnRenamed(old_name, new_name)
return df.show()
   else:
         raise ValueError("'columns' should be a dict, like {'old_name':'new_name', 
                'old_name_one more':'new_name_1'}")
except Exception as e:
      print(e)