如何在Spark/SQL中对datetime列按10分钟间隔聚合均值

需求说明 #

现有包含A、B（字符串格式时间）、C列的DataFrame，需基于B列按每10分钟为时间间隔，聚合计算C列的均值。

输入数据 #

import pandas as pd
import numpy as np

df1 = pd.DataFrame(
    {"A": ["foo", "bar", "foo", "bar", "foo", "bar", "foo", "foo",
          "foo", "bar", "foo", "bar", "foo", "bar", "foo", "foo"],
     "B": ["2013-01-01 01:01:00", "2013-01-01 01:03:00", "2013-01-01 01:06:00", 
           "2013-01-01 01:07:00", "2013-01-01 01:10:00", "2013-01-01 01:13:00", 
           "2013-01-01 01:16:00", "2013-01-01 01:19:00", 
           "2013-01-02 02:01:00", "2013-01-02 02:03:00", "2013-01-02 02:06:00", 
           "2013-01-02 02:07:00", "2013-01-02 02:10:00", "2013-01-02 02:13:00", 
           "2013-01-02 02:16:00", "2013-01-02 02:19:00"],
     "C": np.random.randn(16),
    })

问题分析 #

你尝试使用SQL语法实现，但原生Pandas不支持这种写法，需要使用Pandas内置的时间处理函数来完成需求。

解决方案 #

步骤1：转换时间列类型 #

先将B列从字符串转换为datetime类型，这是时间聚合的前提：

df1['B'] = pd.to_datetime(df1['B'])

步骤2：按10分钟间隔重采样计算均值 #

使用resample方法按10分钟间隔分组，并计算C列的均值。为了匹配预期输出的时间格式（显示区间结束时间），用ceil调整索引：

# 设置B列为索引
df1 = df1.set_index('B')
# 按10分钟重采样，计算均值，调整索引为区间结束时间
result = df1['C'].resample('10T').mean()
result.index = result.index.ceil('10T')
# 重置索引并命名列
result = result.reset_index(name='C_mean')

完整代码 #

import pandas as pd
import numpy as np

df1 = pd.DataFrame(
    {"A": ["foo", "bar", "foo", "bar", "foo", "bar", "foo", "foo",
          "foo", "bar", "foo", "bar", "foo", "bar", "foo", "foo"],
     "B": ["2013-01-01 01:01:00", "2013-01-01 01:03:00", "2013-01-01 01:06:00", 
           "2013-01-01 01:07:00", "2013-01-01 01:10:00", "2013-01-01 01:13:00", 
           "2013-01-01 01:16:00", "2013-01-01 01:19:00", 
           "2013-01-02 02:01:00", "2013-01-02 02:03:00", "2013-01-02 02:06:00", 
           "2013-01-02 02:07:00", "2013-01-02 02:10:00", "2013-01-02 02:13:00", 
           "2013-01-02 02:16:00", "2013-01-02 02:19:00"],
     "C": np.random.randn(16),
    })

# 转换时间列
df1['B'] = pd.to_datetime(df1['B'])
# 重采样计算
df1 = df1.set_index('B')
result = df1['C'].resample('10T').mean()
result.index = result.index.ceil('10T')
result = result.reset_index(name='C_mean')

print(result)

预期输出示例 #

（注：实际均值由随机数生成，以下为格式示例）

B    C_mean
0 2013-01-01 01:10:00  0.123456
1 2013-01-01 01:20:00 -0.789012
2 2013-01-02 02:10:00  0.345678
3 2013-01-02 02:20:00 -0.901234

内容的提问来源于stack exchange，提问作者ferrelwill