环球实时：使用Java和Python进行数据统计和分析

2022-06-25 06:04:59来源：51CTO.COM

Java 和 Python 是当今最流行的两种计算机语言。两者都非常成熟，并提供了工具和技术生态系统，帮助我们解决数据科学领域出现的挑战性问题。每种语言都各有优势，我们要知道什么时候应该使用哪种工具，或者什么时候它们应该协同工作相互补充。

Python 是一种动态类型语言，使用起来非常简单，如果我们不想接触复杂的程序，它肯定是进行复杂计算的首选语言。Python 提供了优秀的库（Pandas、NumPy、Matplotlib、ScyPy、PyTorch、TensorFlow 等）来支持对数据结构或数组的逻辑、数学和科学操作。

Java 是一种非常健壮的语言，具有强类型，因此有更严格的语法规则，所以不易出现程序错误。与Python一样，它也提供了大量的库来处理数据结构、线性代数、机器学习和数据处理（ND4J、Mahout、Spark、Deeplearning4J 等）。

(相关资料图)

本文将介绍如何对大量表格数据进行简单的数据分析，并使用 Java 和 Python 计算一些统计数据。我们可以看到使用各个平台进行数据分析的不同技术，对比它们的扩展方式，以及应用并行计算来提高其性能的可行性。

提出问题

我们要对不同州的一大批城市的价格做一个简单的分析，这里假设有一个包含此信息的 CSV 文件。阅读文件并继续过滤掉一些州，并将剩余的州按城市-州分组以进行一些基本统计。希望能够找到有效执行的解决方案，并且能够随着输入数据规模的增长而有良好的扩展。

数据样本是：

城市	州	基本价格	实际价格
La Jose	PA	34.17	33.19
Preachers Slough	WA	27,46	90.17
Doonan Corners	NY	92.0	162.46
Doonan Corners	NY	97.45	159.46
Castle Rock	WA	162.16	943.21
Marble Rock	IA	97.13	391.49
Mineral	CA	99.13	289.37
Blountville	IN	92.50	557.66
Blountsville	IN	122.50	557.66
Coe	IN	187.85	943.98
Cecilia	KY	92.85	273.61

目的是展示如何使用 Java 和 Python 解决这些类型的问题。该示例非常简单且范围有限，但很容易拓展到更具挑战性的问题。

Java 的方法

首先定义一个封装数据元素的 Java 记录：

record InputEntry(String city, String state, double basePrice, double actualPrice) {}

记录（record）是 JDK 14 中引入的一种新型类型声明。它是定义提供构造函数、访问器、equals 和哈希实现的不可变类的一种简捷方式。

接下来，读取 CVS 文件并将它们增加到一个列表中：

List inputEntries = readRecordEntriesFromCSVFile(recordEntries.csv);

为了按城市和州对输入的元素进行分组，将其定义：

record CityState(String city, String state) {};

使用以下类来封装属于一个组的所有元素的统计信息：

record StatsAggregation(StatsAccumulator basePrice, StatsAccumulator actualPrice) {}

StatsAccumulator是Guava 库的一部分。可以将双精度值集合添加到类中，它会计算基本统计数据，例如计数、平均值、方差或标准差。可以使用StatsAccumulator来获取InputEntry的basePrice和actualPrice的统计数据。

现在我们已经拥有了解决问题的所有材料。Java Streams提供了一个强大的框架来实现数据操作和分析。它的声明式编程风格，对选择、过滤、分组和聚合的支持，简化了数据操作和统计分析。它的框架还提供了一个强大的实现，可以处理大量的（甚至是无限的流），并通过使用并行性、懒惰性和短路操作来高效处理。所有这些特性使Java Streams成为解决这类问题的绝佳选择。实现非常简单：

Map stats = inputEntries.stream().    filter(i -> !(i.state().equals("MN") || i.state().equals("CA"))).collect(        groupingBy(entry -> new CityState(entry.city(), entry.state()),                    collectingAndThen(Collectors.toList(),                                      list -> {StatsAccumulator sac = new StatsAccumulator();                                                sac.addAll(list.stream().mapToDouble(InputEntry::basePrice));                                              StatsAccumulator sas = new StatsAccumulator();                                                sas.addAll(list.stream().mapToDouble(InputEntry::actualPrice));                                               return new StatsAggregation(sac, sas);}                                       )));

在代码的第 2 行，我们使用Stream::filter. 这是一个布尔值函数，用于过滤列表中的元素。可以实现一个 lambda 表达式来删除任何包含“MN”或“CA”状态的元素。

然后继续收集列表的元素并调用Collectors::groupingBy()（第 3 行），它接受两个参数：

一个分类功能，使用CityState记录来做城市和州的分组（第3行）。下游的收集器，包含属于同一<城州>的元素。使用Collectors::collectingAndThen（第 4 行），它采用两个参数分两步进行归约：

·我们使用Collectors::toList（第 4 行），它返回一个收集器，它将属于同一<城州>的所有元素放到一个列表中。

·随后对这个列表进行了整理转换。使用一个lambda函数（第5行至第9行）来定义两个StatsAccumulator（s），在这里分别计算前一个列表中的basePrice和actualPrice元素的统计数据。最后，返回到新创建的包含这些元素的StatsAggregation记录。

正如前文所述，使用Java Streams的优势之一是，它提供了一种简单的机制，可以使用多线程进行并行处理。这允许利用CPU的多核资源，同时执行多个线程。只要在流中添加一个 "parallel"：

Map stats = inputEntries.stream().parallel().

这导致流框架将元素列表细分为多个部分，并同时在单独的线程中运行它们。随着所有不同的线程完成它们的计算，框架将它们串行添加到生成的 Map 中。

在第4行中使用Collectors::groupingByConcurrent而不是Collectors:groupingBy。在这种情况下，框架使用并发映射，允许将来自不同线程的元素直接插入到此映射中，而不必串行组合。

有了这三种可能性，可以检查它们如何执行之前的统计计算（不包括从 CSV 文件加载数据的时间），因为加载量从500万条翻倍到2000万条：

	串行	平行	并行 & GroupByConcurrent
五百万个元素	3.045 秒	1.941 秒	1.436 秒
一千万个元素	6.405 秒	2.876 秒	2.785 秒
两千万个元素	8.507 秒	4.956 秒	4.537 秒

可以看到并行运行大大提高了性能；随着负载的增加，时间几乎减半。使用 GroupByConcurrent 还可额外获得 10% 的收益。

最后，得到结果是微不足道的；例如，要获得印第安纳州 Blountsville 的统计数据，我们只需要：

StatsAggregation aggreg = stateAggr.get(new CityState("Blountsville ", "IN"));System.out.println("Blountsville, IN");System.out.println("basePrice.mean: " + aggreg.basePrice().mean());System.out.println("basePrice.populationVariance: " + aggreg.basePrice().populationVariance());System.out.println("basePrice.populationStandardDeviation: " + aggreg.basePrice().populationStandardDeviation());System.out.println("actualPrice.mean: " + aggreg.basePrice().mean());System.out.println("actualPrice.populationVariance: " + aggreg.actualPrice().populationVariance());System.out.println("actualPrice.populationStandardDeviation: " + aggreg.actualPrice().populationStandardDeviation());

得到的结果：

Blountsville : INbasePrice.mean: 50.302588996763795basePrice.sampleVariance: 830.7527439246837basePrice.sampleStandardDeviation: 28.822781682632293basePrice.count: 309basePrice.min: 0.56basePrice.max: 99.59actualPrice.mean: 508.8927831715211actualPrice.sampleVariance: 78883.35878833274actualPrice.sampleStandardDeviation: 280.86181440048546actualPrice.count: 309actualPrice.min: 0.49actualPrice.max: 999.33

Python的方法

在 Python 中，有几个库可以处理数据统计和分析。其中，Pandas 库非常适合处理大量表格数据，它提供了非常有效的过滤、分组和统计分析方法。

使用 Python 分析以前的数据：

import pandas as pddef group_aggregations(df_group_by):    df_result = df_group_by.agg(        {"basePrice": ["count", "min", "max", "mean", "std", "var"],         "actualPrice": ["count", "min", "max", "mean", "std", "var"]}    )    return df_result  if __name__ == "__main__":    df = pd.read_csv("recordEntries.csv")    excluded_states = ["MN", "CA"]    df_st = df.loc[~ df["state"].isin(excluded_states)]    group_by = df_st.groupby(["city", "state"], sort=False)    aggregated_results = group_aggregations(group_by)

在主要部分，先调用pandas.read_csv()（第 11 行）将文件中用逗号分隔的值加载到 PandasDataFrame中。

在第13行，使用~df["state"].isin(excluded_states)来得到一个Pandas系列的布尔值，使用pandas.loc()来过滤其中不包括的州（MN和CA）。

接下来，在第14行使用DataFrame.groupby()来按城市和州进行分组。结果由group_aggregations()处理，保存每个组的basePrice和actualPrice的统计数据。

在Python中打印结果是非常直接的。IN和Blountsville的结果：

print(aggregated_results.loc["Blountsville", "IN"]["basePrice"])    print(aggregated_results.loc["Blountsville", "IN"]["actualPrice"])

统计数据：

base_price:Name: (Blountsville, IN), dtype: float64count    309.000000min        0.560000max       99.590000mean      50.302589std       28.822782var      830.752744actual_price:Name: (Blountsville, IN), dtype: float64count      309.000000min          0.490000max        999.330000mean       508.892783std        280.861814var      78883.358788

为了并行运行前面的代码，我们必须记住，Python并不像Java那样支持细粒度的锁机制。必须解决好与全局解释器锁（GIL）的问题，无论你有多少个CPU多核或线程，一次只允许一个线程执行。

为了支持并发，我们必须考虑到有一个CPU 密集型进程，因此，最好的方法是使用multiprocessing。所以需要修改代码：

from multiprocessing import Poolimport pandas as pddef aggreg_basePrice(df_group):    ct_st, grp = df_group    return ct_st, grp.basePrice.agg(["count", "min", "max", "mean", "std", "var"])  if __name__ == "__main__":    df = pd.read_csv("recordEntries.csv")    start = time.perf_counter()    excluded_states = ["MN", "CA"]    filtr = ~ df["state"].isin(excluded_states)    df_st = df.loc[filtr]    grouped_by_ct_st = df_st.groupby(["city", "state"], sort=False)    with Pool() as p:        list_parallel = p.map(aggreg_basePrice, [(ct_st, grouped) for ct_st, grouped in grouped_by_ct_st])    print(f"Time elapsed parallel: {round(finish - start, 2)} sec")

和之前一样，使用Pandas groupby()来获得按城市和州分组的数据（第14行）。在下一行，使用多进程库提供的Pool()来映射分组的数据，使用aggreg_basePrice来计算每组的统计数据。Pool()会对数据进行分割，并在几个平行的独立进程中进行统计计算。

正如下面的表格中所示，多进程比串行运行进程慢得多。因此，对于这些类型的问题，不值得使用这种方法。

可以使用另一种并发运行代码 - Modin。Modin提供了一种无缝的方式来并行化你的代码，当你必须处理大量的数据时是非常有用的。将导入语句从import pandas as pd改为import modin.pandas as pd，可以并行运行代码，并利用环境中可能存在的内核集群来加速代码的执行。

下面的表格是刚刚涉及的不同场景的运行时间（和以前一样，不包括从CSV文件中读取数据的时间）：

	串行	多进程	Modin过程
五百万个元素	1.94 秒	20.25 秒	6.99 秒
一千万个元素	4.07 秒	25.1 秒	12.88 秒
两千万个元素	7.62 秒	36.2 秒	25.94 秒