Python的TSFRESH库处理时间太长

Question

我发现TSfresh库是一种将时间序列数据转化为特征的方法。该文档十分出色，似乎非常适合我正在开发的项目。

我想实现TFresh文档快速入门部分中分享的以下代码。看起来相当简单。

from tsfresh import extract_relevant_features
feature_filtered_direct=extract_relevant_features(result,y,column_id=0,column_sort=1)

我的数据包括40万行的传感器数据，每个id有6个传感器，我开始运行代码，17小时后它仍未完成。我想这可能是一个太大的数据集，无法通过相关特征提取器运行，因此我将其缩减到3000，然后进一步缩减到300。没有任何一项措施使代码在一个小时内运行，最终我只好在等待了一个小时左右之后关闭了它。我也尝试了标准的特征提取器。

extracted_features = extract_features(timeseries, column_id="id", column_sort="time")

除尝试TSfresh快速入门部分提供的示例数据集之外，还包括一个非常类似于我的原始数据的数据集，数据点数量减少到了大约相同水平。

有人对这段代码有任何经验吗？如何使其更快运行？我正在使用Anaconda for python 2.7。

更新看起来与多进程有关，因为我在Windows上使用多进程代码需要保护它：

if __name__ == "__main__":
    main()

我添加了一次

if __name__ == "__main__":

    extracted_features = extract_features(timeseries, column_id="id", column_sort="time")

对于我的代码，示例数据有效。但是，我在运行提取相关特征函数和在自己的数据集上运行提取特征模块时仍然遇到一些问题。似乎它仍然运行缓慢。我有一种感觉它与多进程冻结有关，但没有任何错误弹出，所以无法确定。在不到 1% 的数据集上提取特征需要大约 30 分钟的时间。

- Michael Bawol

---

当我在控制台之外运行脚本时，会出现一些有趣的错误消息，这些消息似乎在Anaconda中没有显示出来。这些错误消息不断循环，这可能解释了为什么函数执行时间如此之长（无限）。产生的RuntineError与freeze_support()有关，尝试在引导完成之前启动新进程。不确定这些都是什么意思。 - Michael Bawol

当我在Windows命令提示符中运行时，我遇到了与在当前进程完成引导阶段之前启动新进程有关的相同RuntimeError。但是在Windows上的Ubuntu中，我没有遇到这个错误。 - yellowjacket05

Answer 1

你使用的tsfresh版本是哪个？操作系统是什么？

我们知道一些特征计算器的计算成本非常高。我们无法完全解决这个问题。在未来，我们将实现一些技巧，如缓存，以进一步提高tsfresh的效率。

您是否尝试过使用MinimalFeatureExtractionSettings仅计算基本功能？它只包含诸如最大值、最小值、中位数等基本功能，但应该运行得更快。

 from tsfresh.feature_extraction import MinimalFeatureExtractionSettings
 extracted_features = extract_features(timeseries, column_id="id", column_sort="time", feature_extraction_settings = MinimalFeatureExtractionSettings())

另外，从仓库安装最新版本也是个好主意，可以通过pip install git+https://github.com/blue-yonder/tsfresh实现。我们正在积极开发它，主分支应该包含最新和最新鲜的版本;)

- MaxBenChrist

---

我目前在Windows 10上使用版本0.3.0。我刚刚对大约197,000行数据运行了提取特征模块，用时约为830。当运行特征选择器模块时，似乎需要更长的时间。完整的功能组合在我的分析中非常有用。要将此操作扩展到数百万行，需要相当多的计算能力。 - Michael Bawol

我们刚刚发布了0.4.0版本，其中包含一个ReasonableFeatureExtraction设置对象，它将提取除两个熵特征之外的所有特征。有了这个对象，我能够在i6800k（6核心，4.3 GHz）上处理100,000个长度为1000的时间序列，大约需要3个小时处理1亿行数据。此外，我使用parallelization="per_sample"。 - MaxBenChrist

从11.2版本开始，ReasonableFeatureExtraction已经被恰当地更名为EfficientFCParameters。 - derchambers

Answer 2

语法略有变化（请参见文档），目前的方法如下：

from tsfresh.feature_extraction import EfficientFCParameters, MinimalFCParameters
extract_features(timeseries, column_id="id", column_sort="time", default_fc_parameters=MinimalFCParameters())

或者

extract_features(timeseries, column_id="id", column_sort="time", default_fc_parameters=EfficientFCParameters())

- Guido

Answer 3

自从版本0.15.0以来，我们已经改进了与Apache Spark和dask的绑定。现在可以直接在您通常的dask或Spark计算图中使用tsfresh特征提取。

您可以在tsfresh.convenience.bindings中找到绑定，并在此处查看文档here。例如，对于dask，它可能看起来像这样（假设df是一个dask.DataFrame，例如我们示例中的机器人故障数据帧）

df = df.melt(id_vars=["id", "time"],
             value_vars=["F_x", "F_y", "F_z", "T_x", "T_y", "T_z"],
             var_name="kind", value_name="value")
df_grouped = df.groupby(["id", "kind"])
features = dask_feature_extraction_on_chunk(df_grouped, column_id="id", column_kind="kind",
                                            column_sort="time", column_value="value",
                                            default_fc_parameters=EfficientFCParameters())
                                            # or any other parameter set

使用 dask 或 Spark（或类似的工具）可以帮助处理非常大的数据集 - 可以在多台计算机上分布式地进行工作，因此可以提高内存和速度。当然，我们仍然支持以前的常规分发方式（docu）。

除此之外，还可以将 tsfresh 与任务编排系统（如 luigi）一起运行。您可以创建一个任务： * 仅读取一个id 和 kind 的数据 * 提取特征 * 将结果写入磁盘

然后让 luigi 处理其余的部分。您可以在 我的博客这里 找到此方法的一个可能实现。

- nilpferd1991

Answer 4

我发现，在多核机器上，更好的分配extract_features计算到独立子组（由column_id值标识）的方法是通过使用joblib.Parallel和Loky后端。

例如，您可以针对单个columnd_id值定义特征提取函数，然后应用它。

from joblib import Parallel, delayed
def map_extract_features(df):
    return extract_features(
        timeseries_container=df,
        default_fc_parameters=settings,
        column_id="ID",
        column_sort="DATE",
        n_jobs=1,
        disable_progressbar=True
    ).reset_index().rename({"index":"ID_CONTO"}, axis=1)

out = Parallel(n_jobs=cpu_count()-1)(
    delayed(map_extract_features)(
        my_dataframe[my_dataframe["ID"]==id]
    ) for id in tqdm(my_dataframe["ID"].unique())
)

这种方法所需的内存比直接在extract_features函数中指定column_id要少得多。

- linello