如何在 Pyspark 中处理数据科学问题的异常

Question

如何确定重命名列下面的哪种异常以及如何在 pyspark 中处理它：def rename_columnsName(df, columns):   #provide names in dictionary formatif 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'}")如何通过使用数据集生成异常来测试它。

Answer 1

下面是一个示例，说明如何测试抛出异常的 PySpark 函数。在此示例中，我们将验证如果排序顺序为 则抛出异常"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'"

请注意，该测试正在验证所提供的特定错误消息。

您可以向您的rename_columnsName函数提供无效输入并验证错误消息是否符合您的预期。

Answer 2

我找到了这个问题的解决方案，我们可以像 python 一样在 Pyspark 中处理异常。例如：

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)