0

我在 python 中使用 spark 1.2.0。

我的问题是，在 sql 查询中，如果字段的值为零，我需要用其他值替换它。

我尝试过适用于 1.4.0 但不适用于 1.2.0 的 case /coalese

case when COALESCE("+fld+",0)=0 then "+str(numavgnumlst[0][lock])+" else "+fld+" end.

但是对于 1.2.0，我尝试对 map 做同样的事情

sc = SparkContext(appName="RunModelCCATTR")
sqlContext=SQLContext(sc)
sqlstr="select ..."
nonzerodf=sqlContext.sql(sqlstr)
.....
iifdatadf=nonzerodf.map(lambda candrow:replacezeroforrow(candrow,numavgnumlst))

....
def replacezeroforrow(rowfields,avgvalfields):
   ind=0
   lent=len(rowfields)
   for rowfield in rowfields[4:lent]:
    if rowfield==0:
     rowfields[ind]=avgvalfields[ind]
    ind=ind+1
   return rowfields;

这会引发错误 

TypeError: 'Row' object does not support item assignment

不确定我能做些什么来实现 spark 1.2.0 中的目标。

感谢您的帮助，我认为它现在正在工作..除了列的顺序似乎已经改变..但这可能不是问题。再次感谢

编辑：

这个想法对我帮助很大，需要稍作修改才能解决眼前的问题，-

def replacezeroforrow(rowfields,avgvalfields,dont_replace=[]):
    rdict = rowfields.asDict()
    return Row(dict([(k,avgvalfields[k] if v == 0 and k not in dont_replace else v ) for (k,v) in rdict.items()]))

我修改了原始解决方案以避免“for”出现语法错误。

对方法的调用如下，-

restrictdict=[FieldSet1,FieldSet2,FieldSet3,FieldSet4,modeldepvarcat[0]]
iifdatadf=nonzerodf.map(lambda candrow: replacezeroforrow(candrow,numavgnumlst[0].asDict(),restrictdict))

但是现在我正在尝试访问 iifdatadf，

frstln= iifdatadf.first()
print frstln

我有以下错误

  return "<Row(%s)>" % ", ".join(self)
TypeError: sequence item 0: expected string, dict found

非常感谢帮助。

4

1 回答 1

0

您可以使用字典而不是列表并简单地返回一个新行：

def replacezeroforrow(row, avgvalfields):
    rdict = row.asDict()
    return Row(**{k: avgvalfields[k] if v == 0 and k in avgvalfields
        else v for (k, v) in rdict.items()})

用法：

>>> r1 = Row(fld1="a", fld2=99, fld3=0, fld4=0)
>>> avgvalfields = {'fld3': 3, 'fld4': 1}
>>> replacezeroforrow(r1, avgvalfields)
Row(fld1='a', fld2=99, fld3=3, fld4=1)
于 2015-08-28T14:52:34.990 回答