我有以下代码：
raw_data = pd.DataFrame({'username':list('ab')*10, 'user_agent': list('cdef')*5, 'method':['POST'] * 20, 'dst_port':[80]*20, 'dst':['1.1.1.1']*20})
past = pd.DataFrame({'user_agent':list('cde'), 'percent':[0.3, 0.3, 0.4]})
past = past.set_index('user_agent')
import dask.dataframe as dd
dask_raw = dd.from_pandas(raw_data, npartitions=4)
dask_past = dd.from_pandas(past, npartitions=4)
merged_raw = dask_raw.merge(dask_past, how='left', left_on='user_agent', right_index=True)
Compute on merged_raw提供此表单的数据帧：
Out[20]:
dst dst_port method user_agent username percent
12 1.1.1.1 80 POST c a 0.3
16 1.1.1.1 80 POST c a 0.3
8 1.1.1.1 80 POST c a 0.3
0 1.1.1.1 80 POST c a 0.3
4 1.1.1.1 80 POST c a 0.3
10 1.1.1.1 80 POST e a 0.4
11 1.1.1.1 80 POST f b NaN
13 1.1.1.1 80 POST d b 0.3
14 1.1.1.1 80 POST e a 0.4
15 1.1.1.1 80 POST f b NaN
17 1.1.1.1 80 POST d b 0.3
18 1.1.1.1 80 POST e a 0.4
19 1.1.1.1 80 POST f b NaN
5 1.1.1.1 80 POST d b 0.3
6 1.1.1.1 80 POST e a 0.4
7 1.1.1.1 80 POST f b NaN
9 1.1.1.1 80 POST d b 0.3
1 1.1.1.1 80 POST d b 0.3
2 1.1.1.1 80 POST e a 0.4
3 1.1.1.1 80 POST f b NaN
计算功能：
grouped_by_df = merged_raw.groupby(['username', 'dst', 'dst_port'])
feature_one = grouped_by_df.apply(lambda x: 'POST' in x.values).to_frame(name='feature_one')
feature_two = grouped_by_df.percent.min()
feature_two = feature_two.fillna(0)
feature_two = feature_two.to_frame(name='feature_two')
features_three = grouped_by_df.method.apply(lambda x: 'CONNECT' in x.values).to_frame(name='feature_three')
features = feature_one.merge(feature_two, left_index=True, right_index=True, how='left')
features.compute()
feature_one feature_two
username dst dst_port
a 1.1.1.1 80 True 0.3
b 1.1.1.1 80 True 0.3
features_full = features.merge(features_three, how='left', right_index=True, left_index=True)
features_full.compute()
结果是：
Out[53]:
Empty DataFrame
Columns: [feature_one, feature_two, feature_three]
Index: []
但是
features_three具有值并且与特征相同feature_three.compute()
username dst dst_port
a 1.1.1.1 80 False
b 1.1.1.1 80 False
为什么dask返回空数据帧？
最佳答案
这并不能完全解决您的问题，但是如果我先计算合并的原始数据帧，我会得到以下结果。如果我注释合并的_raw.compute（）命令，就会得到一条错误消息。我想知道您是否可以一直使用pandas并使用dask延迟函数进行并行计算。
import dask.dataframe as dd
import pandas as pd
raw_data = pd.DataFrame({'username':list('ab')*10, 'user_agent': list('cdef')*5, 'method':['POST'] * 20, 'dst_port':[80]*20, 'dst':['1.1.1.1']*20})
past = pd.DataFrame({'user_agent':list('cde'), 'percent':[0.3, 0.3, 0.4]})
past = past.set_index('user_agent')
dask_raw = dd.from_pandas(raw_data, npartitions=4)
dask_past = dd.from_pandas(past, npartitions=4)
merged_raw = dask_raw.merge(dask_past, how='left', left_on='user_agent', right_index=True)
merged_raw = merged_raw.compute()
grouped_by_df = merged_raw.groupby(['username', 'dst', 'dst_port'])
feature_one = grouped_by_df.apply(lambda x: 'POST' in x.values).to_frame(name='feature_one')
feature_two = grouped_by_df.percent.min()
feature_two = feature_two.fillna(0)
feature_two = feature_two.to_frame(name='feature_two')
features_three = grouped_by_df.method.apply(lambda x: 'CONNECT' in x.values).to_frame(name='feature_three')
features = feature_one.merge(feature_two, left_index=True, right_index=True, how='left')
features_full = features.merge(features_three, how='left', right_index=True, left_index=True)
features_full
Out[85]:
feature_one feature_two feature_three
username dst dst_port
a 1.1.1.1 80 True 0.3 False
b 1.1.1.1 80 True 0.3 False
# Error message when the merged_raw.compute() command is commented:
C:/CODE/apostolos.py:23: UserWarning: `meta` is not specified, inferred from partial data. Please provide `meta` if the result is unexpected.
Before: .apply(func)
After: .apply(func, meta={'x': 'f8', 'y': 'f8'}) for dataframe result
or: .apply(func, meta=('x', 'f8')) for series result
feature_one = grouped_by_df.apply(lambda x: 'POST' in x.values).to_frame(name='feature_one')
C:/CODE/apostolos.py:31: UserWarning: `meta` is not specified, inferred from partial data. Please provide `meta` if the result is unexpected.
Before: .apply(func)
After: .apply(func, meta={'x': 'f8', 'y': 'f8'}) for dataframe result
or: .apply(func, meta=('x', 'f8')) for series result
features_three = grouped_by_df.method.apply(lambda x: 'CONNECT' in x.values).to_frame(name='feature_three')