是否可以在 map() 中删除 Pyspark 行？

Question

在 Pyspark 中执行映射函数时，我经常想删除映射函数失败的数据（在此示例中，转换为 xml）。我想知道在映射步骤中是否有一种干净的方法可以做到这一点？

返回空白的明显解决方案仍然在RDD中留下一个对象，例如。

### **** skip pyspark boilerplate ****

### function defs
from lxml import etree as ET
def return_valid_xml(one_line_input):
    try:
        root = ET.fromstring(one_line_input)
        return root
    except:
        return

### code that returns stuff for every line of input
valid_xml_data = someDataStrings.map(lambda x: return_valid_xml(x))

想出一个聪明的过滤器是在浪费我的时间，而像 ET.fromstring() return true 上的 try/except 之类的愚蠢过滤器是在浪费计算时间，因为我解析了两次 XML。

Answer 1

您可以使用flatMap 并在失败时返回一个空列表：

def return_valid_xml(one_line_input):
    try:
        root = ET.fromstring(one_line_input)
        return [root]
    except:
        return []

valid_xml_data = someDataStrings.flatMap(return_valid_xml)

您也可以只传入return_valid_xml，而不是定义lambda 函数。