如何使用 Python、Elasticsearch 和 Kibana 分析数据

引言

在这篇博文中，我将演示使用 Elasticsearch、Python 和 Kibana 加载、分析和可视化数据的过程。虽然示例将围绕波士顿凯尔特人队的数据展开，但这些技术和概念可以应用于其他数据集。您可以查看这篇博文的完整代码。

选择使用波士顿凯尔特人队的数据源于我个人的经历。大约一年前，我经历了一段压力很大的时期，最终去看了场篮球比赛。在那段时间里，我能够以以前从未有过的方式放松身心并集中注意力。我开始关注的第一支球队就是波士顿凯尔特人队。波士顿凯尔特人队是一支独特的球队，虽然在本赛季他们经常位列 NBA 实力榜前列，但他们并非总能在许多核心指标上领先联盟。

通过数据可视化，我可以更深入地了解这支球队，回答一些关于它的关键问题，并更好地分析这个赛季。

使用 Python、Elasticsearch 和 Kibana 分析数据的步骤

先决条件

本教程使用 Elasticsearch 8.11 版；如果您是新手，请查看我们关于Elasticsearch和Kibana的快速入门指南。
如果您尚未在计算机上安装 Python，请下载最新版本。此示例使用 Python 3.12.1。
您将使用nba_api包获取有关波士顿凯尔特人队的最新统计数据，Jupyter Notebooks和Elasticsearch Python 客户端。在测试此代码时，除非我安装了pandas，否则我遇到了错误，因为 nba_data 会创建 pandas DataFrame。

要安装这些包，您可以运行以下命令。

pip3 install nba_api jupyter elasticsearch pandas

您需要加载一个 Jupyter Notebook 以交互方式处理您的数据。为此，您可以在终端中运行以下命令。

jupyter notebook

在右上角，您可以选择“新建”以创建新的 Jupyter Notebook。

步骤 1：解析和清理数据

第一步将是连接到 NBA 数据并将这些数据加载到 Elasticsearch 中。您首先需要导入所需的库。在此示例中，您将使用 nba_api 的静态团队数据来获取有关波士顿凯尔特人队的信息。leaguegamefinder 端点允许您获取信息。要连接到 Elastic，您将使用 Elasticsearch Python 客户端 elasticsearch。要通过交互式提示输入您的密码，您将使用 getpass。

要加载这些包，您可以使用以下代码

from nba_api.stats.static import teams
from nba_api.stats.endpoints import leaguegamefinder
from elasticsearch import Elasticsearch, helpers
from getpass import getpass

您需要从 NBA 球队静态数据集中获取球队数据，该数据集为每个球队分配了一个 ID。您可以使用列表推导式查找缩写为 BOS（波士顿）的球队。获取完整的凯尔特人队对象后，您可以将其缩小到仅 ID，然后使用它查找比赛数据。

nba_teams = teams.get_teams()
celtics = [team for team in nba_teams if team['abbreviation'] == 'BOS'][0]
celtics_id = celtics['id']

现在，您可以使用凯尔特人队的 ID 获取该球队的所有可用比赛数据。您可以使用 .head() 方法查看前五个结果以确保数据已正确加载。

celtics_games = leaguegamefinder.LeagueGameFinder(team_id_nullable=celtics_id)
games = celtics_games.get_data_frames()[0]
games.head()

在处理这些数据时，我注意到该年份的数据包括季前赛数据。因此，我使用赛季日期将数据缩小到当前赛季。在 Jupyter Notebook 中，您可以调用 current_season 以查看完整的 DataFrame。

current_season = games.loc[(games['GAME_DATE'] >= '2023-10-24') & (games['GAME_DATE'] <= '2024-06-20')]
current_season

由于空值在将数据加载到 Elasticsearch 时可能会导致问题，因此您可以仔细检查这些数据是否没有空值。以下行将返回一个布尔值，让您知道您的数据是否包含任何空值。由于此数据集返回 False，因此它没有空值，所以我们无需进行进一步清理。

current_season.isnull().values.any()

步骤 2：将数据加载到 Elasticsearch 中

解析 NBA 数据并完成数据清理后，您可以为 Elastic Cloud ID 和 Elastic API 密钥创建变量。使用 getpass，您可以安全地输入您的凭据。

elastic_cloud_id = getpass("Elastic Cloud ID: ")
elastic_api_key = getpass("API Key: ")

输入凭据后，您可以使用 elasticsearch 客户端连接到 Elasticsearch。

es = Elasticsearch(cloud_id=elastic_cloud_id, api_key=elastic_api_key)

在将数据加载到 Elastic 之前，您必须创建一个索引。您可以为当前赛季创建一个索引。

es.indices.create(index="boston_celtics_current_season")

您可以创建一个函数，将当前赛季的数据加载到 Elasticsearch 中。每场比赛都被视为一个文档。

timeframe = 'boston_celtics_current_season'

def doc_generator(df, timeframe):
    for index, document in df.iterrows():
        yield {
            "_index": timeframe, 
            "_id": f"{document['GAME_ID']}",
            "_source": document.to_dict(),
        }

Python 客户端的 helpers 功能允许您有效地将包含当前赛季比赛数据的 DataFrame 上传到 Elasticsearch。通过调用您刚刚创建的 doc_generator 函数，您可以将 DataFrame 转换为文档。

helpers.bulk(es, doc_generator(current_season, timeframe))

步骤 3：使用 Elasticsearch 编写查询

现在您的数据已加载，您可以开始使用 Elasticsearch 编写查询以了解有关波士顿凯尔特人队在本赛季的表现的更多信息。首先，您可以创建一个查询以查看他们在本赛季迄今为止赢得了多少场比赛，并返回获胜次数作为结果。

search_query = {
    "query": {
        "match": {
            "WL": "W"
        }
    }
}

games_won = es.count(index="boston_celtics_current_season", body=search_query)

在处理复杂数据集时，编写句子来帮助解释数据集有时很有帮助。以下是一个示例，说明波士顿凯尔特人队在本赛季赢得了多少场比赛。

print(f"The Celtics won {games_won['count']} games this season so far.")

输出应如下所示

The Celtics won 29 games this season so far.

在体育运动中，连胜是指一个团队或个人连续赢得或输掉一系列比赛或事件。连胜很重要，因为它们反映了一段时间的出色表现（连胜）或充满挑战的阶段（连败）。在分析一个团队的表现时，检查他们有多少连胜通常很有帮助。您可以创建一个查询，允许您按比赛日期对胜负进行排序。

streak_query = {
  "size": 1000,  
  "sort": [
    {
      "GAME_DATE": {
        "order": "asc"
      }
    }
  ],
  "_source": ["GAME_DATE", "WL"]
}

您可以使用 es.search() 方法根据上述查询创建搜索。

streak_search = es.search(
    index="boston_celtics_current_season",
    body=streak_query)

以下代码创建了一个包含比赛日期和比赛结果的 JSON 对象。

gs = [hit['_source'] for hit in streak_search['hits']['hits']]

要查看本赛季的前五名连胜，您可以创建一个包含每个连胜的字典并对其进行相应的排序。

streaks = []
current_streak = 1
for i in range(1, len(gs)):
    if gs[i]['WL'] == gs[i-1]['WL']:
        current_streak += 1
    else:
        streaks.append((gs[i-1]['WL'], current_streak))
        current_streak = 1


streaks.append((gs[-1]['WL'], current_streak))
top_streaks = sorted(streaks, key=lambda x: x[1], reverse=True)[:5]
top_streaks

步骤 4：使用 Kibana 创建仪表板

虽然我们可以继续编写查询以了解有关波士顿凯尔特人队的更多信息，但创建仪表板是更有效地从数据中获取见解的方法。

在创建仪表板之前，您需要创建一个数据视图以确保 Kibana 可以访问 Elasticsearch 索引中的数据。对于您的数据视图，您需要为其命名，选择要可视化的索引或表示多个索引的模式，并提供一个时间戳字段，以便您可以创建基于时间可视化。

创建数据视图后，您可以开始创建仪表板。在“分析”标题下，选择“仪表板”，然后点击“创建仪表板”。

一个很好的入门可视化是为仪表板创建标题可视化。您可以选择文本可视化和 Markdown 以将图像添加到标题中。

# ![image](https://1000logos.net/wp-content/uploads/2016/10/Boston-Celtics-Logo.png) How are the Boston Celtics performing this season?

要了解凯尔特人队是否赢的比赛多于输的比赛，您可以创建华夫饼图来说明在本赛季的这一点上，凯尔特人队赢的比赛多于输的比赛。

您可以在此处查看此图表配置

保持助攻次数多于失误次数是团队有效控球的重要指标。随附的可视化（截至本文发表之日）清楚地说明了团队在这方面表现出色，展示了熟练的控球分配和团队合作。

此可视化的配置如下所示

篮球比赛中的正负值表示波士顿凯尔特人队比其他球队多得了多少分，这个数据通常用于解释一支球队对比赛的影响。较高的正分值表明，当该球队比赛时，他们在得分或防止对手进球/得分方面表现良好。较高的负分值则表示相反的情况——该球队往往得分较少。赛季初出现了一个异常值，有一场比赛凯尔特人队比另一支球队多得了50多分，但这后来随着时间的推移逐渐趋于正常。最近的一场比赛（在撰写本文时），对阵密尔沃基雄鹿队，也同样是一个异常值。

以下是上述可视化的配置。

要了解凯尔特人队投篮的频率，您可以创建一些顶部线条，包括

投篮命中率的平均值是多少？
三分线投篮命中率的平均值是多少？
罚球命中率的平均值是多少？

这些顶部线条的配置在 Kibana 中如下所示

投篮命中率顶部线条的配置

三分线投篮命中率顶部线条的配置

罚球命中率顶部线条的配置

最终的可视化检查了凯尔特人队的得分是否随着抢断和盖帽次数的增加而增加。颜色代表盖帽平均值加上抢断平均值。随着盖帽和抢断次数的增加，绿色会变深。然而，数据中缺乏明显的模式表明，这些防守动作与其整体得分之间没有显著的相关性。

此可视化的配置应如下所示

结论

通过以这种方式可视化数据，您可以稳健地与您的数据交互并得出更多见解。请务必查看此博文文章的完整代码。下一步，创建一个数据管道，以编程方式将数据导入您的仪表板，或利用一些机器学习功能，例如异常检测。您还可以通过添加凯尔特人队的历史数据或将凯尔特人队与 NBA 中的其他球队进行比较来扩展此数据集。我们希望您能继续使用 Python、Elasticsearch 和 Kibana。与往常一样，如果您需要此博文文章启发您构建任何内容，或者您对我们的Discuss 论坛和社区 Slack 频道有任何疑问，请告诉我们。

想要获得 Elastic 认证？了解下一次Elasticsearch 工程师培训何时开始！

Elasticsearch 拥有众多新功能，可帮助您为您的用例构建最佳搜索解决方案。深入了解我们的示例笔记本以了解更多信息，开始免费云试用，或立即在您的本地机器上试用 Elastic。