PySpark UDF生成的DF无法显示“值错误：”MyColumn“名称不在列表中”

Question

data.show()
+---------------+--------------------+--------------------+
|       features|                meta|           telemetry|
+---------------+--------------------+--------------------+
|   [seattle, 3]|[seattle, 3, 5344...|[[47, 1, 27, 92, ...|
|     [miami, 1]|[miami, 1, 236881...|[[31, 84, 24, 67,...|
|     [miami, 3]|[miami, 3, 02f4ca...|[[84, 5, 4, 93, 2...|
|   [seattle, 3]|[seattle, 3, ec48...|[[43, 16, 94, 93,...|
|   [seattle, 1]|[seattle, 1, 7d19...|[[70, 22, 45, 74,...|
|[kitty hawk, 3]|[kitty hawk, 3, d...|[[46, 15, 56, 94,...|

您可以从以下链接下载生成的.json示例:https://aiaccqualitytelcapture.blob.core.windows.net/streamanalytics/2019/08/21/10/0_43cbc7b0c9e845a187ce182b46eb4a3a_1.json？

特别是，您可以看到其中每一个中的实际数据实际上都是一个字典：我们感兴趣的“features”列的形式如下：{“factory_id”：“西雅图”,“line_id”：“3”}

我试图通过经典的函数方法将特性中的数据编码为one_hot。

见下文：

def one_hot(value, categories_list):
  num_cats = len(categories_list)
  one_hot = np.eye(num_cats)[categories_list.index(value)]
  return one_hot

def one_hot_features(row, feature_keys, u_features):
  """
  feature_keys must be sorted.
  """
  cur_key = feature_keys[0]
  vector = one_hot(row["features"][cur_key], u_features[cur_key])
  for i in range(1, len(feature_keys)):
    cur_key = feature_keys[i]
    n_vector = one_hot(row["features"][cur_key], u_features[cur_key])
    vector = np.concatenate((vector,  n_vector), axis=None)
  return vector

本例中的feature_keys和u_features包含以下数据：

feature_keys = ['factory_id', 'line_id']

u_features = {'factory_id': ['kitty hawk', 'miami', 'nags head', 'seattle'], 'line_id': ['1', '2', '3']}

def calc_onehot_udf(feature_keys, u_features):
  return udf(lambda x: one_hot_features(x, feature_keys, u_features))

n_data = data.withColumn("hot_feature", calc_onehot_udf(feature_keys, 
u_features)( col("features") ))

n_data.show()

理想的输出是一个新的dataframe，其列为：“hot_features”，其中包含features列中的1维一个hot encoded数组。

Answer 1

结果发现有几个关键问题：

1. 您应该或必须在UDF中指定返回类型。在本例中，它是ArrayType(FloatType())
2. 没有从one_hot_features返回nd数组，而是调用了vectors.tolist（）
3. 传递col（“features”）逐行发送features列内的实际值，而不是实际行数据；因此，像最初那样调用row[“features”]是不正确的，因为没有访问器，因为我已经有了该行的值。因此，我将第一个参数重命名为“features_val”，而不是“row”，以更好地反映预期的输入。

下面是one_hot_features的新代码。

def one_hot_features(features_val, feature_keys, u_features):
  cur_key = feature_keys[0]
  vector = one_hot(features_val[cur_key], u_features[cur_key])
  for i in range(1, len(feature_keys)):
    cur_key = feature_keys[i]
    n_vector = one_hot(features_val[cur_key], u_features[cur_key])
    vector = np.concatenate((vector,  n_vector), axis=None)
  return vector.tolist()

def calc_onehot_udf(feature_keys, u_features):
  return udf(lambda x: one_hot_features(x, feature_keys, u_features), 
ArrayType(FloatType()))

n_data = data.withColumn("hot_feature", calc_onehot_udf(feature_keys, 
u_features)(col("features")))
n_data.show()