我有一个糟糕的HTTPD访问日志,只想跳过“糟糕”的行。
在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一样)?
非常感谢!