相关疑难解决方法(0)

我有一个糟糕的HTTPD access_log,只想跳过"糟糕"的行.

在scala中,这很简单:

import scala.util.Try

val log = sc.textFile("access_log")

log.map(_.split(' ')).map(a => Try(a(8))).filter(_.isSuccess).map(_.get).map(code => (code,1)).reduceByKey(_ + _).collect()

对于python我通过使用"lambda"表示法明确定义一个函数来得到以下解决方案:

log = sc.textFile("access_log")

def wrapException(a):
    try:
        return a[8]
    except:
        return 'error'

log.map(lambda s : s.split(' ')).map(wrapException).filter(lambda s : s!='error').map(lambda code : (code,1)).reduceByKey(lambda acu,value : acu + value).collect()

在pyspark中有更好的方法(例如在Scala中)吗？

非常感谢!