python - 如何根据某些条件更新字典的 PySpark RDD

Question

首先，我了解 RDD 的持久数据结构和不变性的概念。更新是我能想到的唯一词:)

我的问题是:

给定一个字典(或 Row 对象)的 RDD，我如何循环/映射并在该 RDD 上应用一些转换登录，并接收应用了这些转换的新 RDD。示例:

给定一个包含字典的 RDD:

fbb = sc.parallelize(
    [{'amount_gbp': -43.33,
      'balance_gbp': 57.08,
      'type': 'GED',
      'id': 961690979,
      'settled_jrnl_cr_datetime': u'(null)',
      'virtual_cash_balance': 0,
      'virtual_debt_balance': 0},
     {'amount_gbp': 17.08,
      'balance_gbp': 40.0,
      'type': 'OIP',
      'id': 962182953,
      'settled_jrnl_cr_datetime': u'(null)',
      'virtual_cash_balance': 0,
      'virtual_debt_balance': 0}])

我尝试应用该功能:

def update_virtual_cash_balance(x):
    x.update({'virtual_cash_balance': x['amount_gbp'] + x['balance_gbp']}) if x['type'] == 'GED' else x

   fbb.map(lambda x: update_virtual_cash_balance(x)).collect()

预期:

[{'amount_gbp': -43.33,
  'balance_gbp': 57.08,
  'type': 'GED',
  'id': 961690979,
  'settled_jrnl_cr_datetime': u'(null)',
  'virtual_cash_balance': 13.75,
  'virtual_debt_balance': 0},
 {'amount_gbp': 17.08,
  'balance_gbp': 40.0,
  'type': 'OIP',
  'id': 962182953,
  'settled_jrnl_cr_datetime': u'(null)',
  'virtual_cash_balance': 0,
  'virtual_debt_balance': 0}]

但是得到了:

Out[411]: [None, None]

任何对我的误解的帮助都会很棒。

Answer 1

• update_virtual_cash_balance 不会返回任何内容，因此您会得到 None
• update 方法不会返回任何内容，因此即使 update_virtual_cash_balance 返回值，您也会得到 None
• 您不应就地修改数据。 RDD 是不可变的，改变数据可能会产生不良影响。

尝试:

def update_virtual_cash_balance(x):
    if x['type'] == 'GED':
        z = x.copy()  # shallow copy should be enough here
        z.update({'virtual_cash_balance': x['amount_gbp'] + x['balance_gbp']}
        return z
    return  x