Elasticsearch:让你的 Elasticsearch 索引与 Python 和 Google Cloud Platform 功能保持同步

作者:来自 Elastic Garson

Elasticsearch 内的索引 (index) 是你可以将数据存储在文档中的位置。 在使用索引时,如果你使用的是动态数据集,数据可能会很快变旧。 为了避免此问题,你可以创建一个 Python 脚本来更新索引,并使用 Google Cloud Platform (GCP) 的 Cloud Functions 和 Cloud Scheduler 进行部署,以便自动保持索引最新。

为了使索引保持最新,你可以首先设置一个 Jupyter Notebook 在本地进行测试,并创建一个脚本框架,该脚本框架将在出现新信息时更新你的索引。 你可以调整脚本以使其更具可重用性并将其作为云函数(Cloud function)运行。 使用 Cloud Scheduler,你可以将 Cloud Function 中的代码设置为使用 cron 类型格式按计划运行。

先决条件

  • 本示例使用 Elasticsearch 版本 8.12; 如果你是新手,请查看我们的 Elasticsearch 快速入门。
  • 如果你的计算机上尚未安装 Python,请下载最新版本。 此示例使用 Python 3.12.1。
  • NASA API 的 API 密钥。
  • 你将使用 Requests 包连接到 NASA API,使用 Pandas 操作数据,使用 Elasticsearch Python 客户端将数据加载到索引中并使其保持最新,并使用 Jupyter Notebooks 在测试时以交互方式处理数据。 你可以运行以下行来安装这些必需的软件包:
pip3 install requests pandas elasticsearch notebook

加载和更新你的数据集

在 GCP 内运行更新脚本之前,你需要上传数据并测试用于保持脚本更新的流程。 你将首先从 API 连接到数据,将其保存为 Pandas DataFrame,连接到 Elasticsearch,将 DataFrame 上传到索引中,检查索引上次更新的时间,并在有新数据可用时更新索引。 你可以在此搜索实验室笔记本中找到本节的完整代码。

加载你的数据

让我们开始使用 Jupyter Notebook 进行本地测试,以交互方式处理你的数据。 为此,你可以在终端中运行以下命令。

jupyter notebook

在右上角,你可以选择 “New” 来创建新的 Jupyter Notebook。

首先,你需要导入将要使用的包。 你将导入之前安装的所有软件包,以及用于处理 API 密钥等机密的 getpass 和用于处理日期对象的 datetime。

import requests
from getpass import getpass
import pandas as pd
from datetime import datetime, timedelta
from elasticsearch import Elasticsearch, helpers

你将使用的数据集是近地天体 Web 服务 (NeoWs),这是一种提供近地小行星信息的 RESTful Web 服务。 该数据集可让你根据小行星最接近地球的日期搜索小行星、查找特定小行星并浏览整个数据集。

通过以下函数,你可以连接到 NASA 的 NeoWs API,获取过去一周的数据,并将响应转换为 JSON 对象。

def connect_to_nasa():url = "https://api.nasa.gov/neo/rest/v1/feed"nasa_api_key = getpass("NASA API Key: ")today = datetime.now()params = {"api_key": nasa_api_key,"start_date": today - timedelta(days=7),"end_date": datetime.now(),}return requests.get(url, params).json()

现在,你可以将 API 调用的结果保存到名为 “response” 的变量中。

要将 JSON 对象转换为 pandas DataFrame,你必须将嵌套对象规范化为一个 DataFrame,并删除包含嵌套 JSON 的列。

def create_df(response):all_objects = []for date, objects in response["near_earth_objects"].items():for obj in objects:obj["close_approach_date"] = dateall_objects.append(obj)df = pd.json_normalize(all_objects)return df.drop("close_approach_data", axis=1)

要调用此函数并查看数据集的前五行,你可以运行以下命令:

df = create_df(response)
df.head()

连接到 Elasticsearch

你可以通过提供 Elastic Cloud ID 和 API 密钥进行身份验证,从 Python 客户端访问 Elasticsearch。

def connect_to_elastic():elastic_cloud_id = getpass("Elastic Cloud ID: ")elastic_api_key = getpass("Elastic API Key: ")return Elasticsearch(cloud_id=elastic_cloud_id, api_key=elastic_api_key)

现在,你可以将连接函数的结果保存到名为 es 的变量中。

es = connect_to_elastic()

Elasticsearch 中的索引是数据的主要容器。 你可以将索引命名为 asteroid_data_set。

index_name = "asteroid_data_set"
es.indices.create(index=index_name)

你返回的结果将如下所示:

ObjectApiResponse({'acknowledged': True, 'shards_acknowledged': True, 'index': 'asteroids_data'})

现在,你可以创建一个辅助函数,该函数允许你将 DataFrame 转换为正确的格式以上传到索引中。

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

接下来,你可以将 DataFrame 的内容批量上传到 Elastic,调用您刚刚创建的辅助函数。

helpers.bulk(es, doc_generator(df, index_name))

你应该得到类似于以下内容的结果,它告诉你已上传的行数:

(146, [])

你最后一次更新数据是什么时候?

将数据上传到 Elasticsearch 后,你可以检查上次更新索引的时间并设置日期格式,以便它可以与 NASA API 配合使用。

def updated_last(es, index_name):query = {"size": 0,"aggs": {"last_date": {"max": {"field": "close_approach_date"}}},}response = es.search(index=index_name, body=query)last_updated_date_string = response["aggregations"]["last_date"]["value_as_string"]datetime_obj = datetime.strptime(last_updated_date_string, "%Y-%m-%dT%H:%M:%S.%fZ")return datetime_obj.strftime("%Y-%m-%d")

你可以将索引上次更新的日期保存到变量中并打印出该日期。

last_update_date = updated_last(es, index_name)
print(last_update_date)

更新你的数据

现在,你可以创建一个函数来检查自上次更新索引和当前日期以来是否有任何新数据。 如果对象有效并且数据不为空,它将更新索引并让你知道是否没有新数据要更新,或者 DataFrame 是否返回 None 类型,表明可能存在问题。

def update_new_data(df, es, last_update_date, index_name):if isinstance(last_update_date, str):last_update_date = datetime.strptime(last_update_date, "%Y-%m-%d")last_update_date = pd.Timestamp(last_update_date).normalize()if not df.empty and "close_approach_date" in df.columns:df["close_approach_date"] = pd.to_datetime(df["close_approach_date"])today = pd.Timestamp(datetime.now().date()).normalize()if df is not None and not df.empty:update_range = df.loc[(df["close_approach_date"] > last_update_date)& (df["close_approach_date"] < today)]if not update_range.empty:helpers.bulk(es, doc_generator(update_range, index_name))else:print("No new data to update.")else:print("The DataFrame is None.")

如果 DataFrame 是有效对象,它将调用你编写的函数并更新索引(如果适用)。 它还会打印出索引上次更新的日期,以帮助你在需要时进行调试。 如果没有,它会告诉你可能有问题。

try:if df is None:raise ValueError("DataFrame is None. There may be a problem.")update_new_data(df, es, last_update_date, index_name)print(updated_last(es, index_name))
except Exception as e:print(f"An error occurred: {e}")

保持索引最新

现在你已经创建了用于本地测试的框架,你可以设置一个环境,你可以每天运行脚本来检查是否有任何新数据可用并相应地更新索引。

创建云函数

你现在已准备好部署云功能。 为此,你需要选择环境作为第二代函数,命名你的函数,然后选择云区域。 你还可以将其绑定到 Cloud Pub/Sub 触发器,并选择创建新主题(如果你尚未创建)。 你可以在 GitHub 上查看本节的完整代码。

创建 Pub/Sub 主题

创建新主题时,你可以命名主题 ID 并选择使用 Google 管理的加密密钥进行加密。

设置 Cloud Function 的环境变量

在 “Runtime environment variables” 下方,你可以添加 NASA_API_KEY、ELASTIC_CLOUD_ID 和 ELASTIC_API_KEY 的环境变量。 你需要将它们保存为原始值,并且不带单引号。 因此,如果你之前在终端中输入了 “xxxlsdgzxxxxx” 值,你会希望它是 xxxlsdgzxxxxx。

调整你的代码并将其添加到你的云函数

输入环境变量后,你可以按 “下一步(next)” 按钮,这将带你进入代码编辑器。 你需要选择 Python 3.12.1 的运行时或匹配你正在使用的 Python 版本。 之后,将入口点更新为 update_index。 入口点的作用与 Python 中的 main 函数类似。

你将希望使用 os 来执行更自动化的过程,而不是使用 getpass 来获得环境变量(账号信息)。 示例如下所示:

elastic_cloud_id = os.getenv("ELASTIC_CLOUD_ID")
elastic_api_key = os.getenv("ELASTIC_API_KEY")

你需要调整脚本的顺序,以使函数首先连接到 Elasticsearch。 之后,你将想知道索引上次更新的时间,连接到你正在使用的 NASA API,将其保存到 DataFrame,并加载可能可用的任何新数据。

你可能会注意到底部有一个名为 update_index 的新函数,它将你的代码联系在一起。 在此函数中,你定义索引的名称,连接到 Elasticsearch,计算出索引的最后更新日期,连接到 NASA API,将结果保存到数据框中,并在需要时更新索引。 要指示入口点函数是云事件,你可以使用装饰器 @functions_framework.cloud_event 来表示它。

@functions_framework.cloud_event
def update_index(cloud_event):index_name = "asteroid_data_set"es = connect_to_elastic()last_update_date = updated_last(es, index_name)print(last_update_date)response = connect_to_nasa(last_update_date)df = create_df(response)if df is not None:update_new_data(df, es, last_update_date, index_name)print(updated_last(es, index_name)) else:print("No new data was retrieved.")

这是完整更新的代码示例:

import functions_framework
import requests
import os
import pandas as pd
from datetime import datetime
from elasticsearch import Elasticsearch, helpersdef connect_to_elastic():elastic_cloud_id = os.getenv("ELASTIC_CLOUD_ID")elastic_api_key = os.getenv("ELASTIC_API_KEY")return Elasticsearch(cloud_id=elastic_cloud_id, api_key=elastic_api_key)def connect_to_nasa(last_update_date):url = "https://api.nasa.gov/neo/rest/v1/feed"nasa_api_key = os.getenv("NASA_API_KEY")params = {"api_key": nasa_api_key,"start_date": last_update_date,"end_date": datetime.now(),}return requests.get(url, params).json()def create_df(response):all_objects = []for date, objects in response["near_earth_objects"].items():for obj in objects:obj["close_approach_date"] = dateall_objects.append(obj)df = pd.json_normalize(all_objects)return df.drop("close_approach_data", axis=1)def doc_generator(df, index_name):for index, document in df.iterrows():yield {"_index": index_name,"_id": f"{document['close_approach_date']}","_source": document.to_dict(),}def updated_last(es, index_name):query = {"size": 0,"aggs": {"last_date": {"max": {"field": "close_approach_date"}}},}response = es.search(index=index_name, body=query)last_updated_date_string = response["aggregations"]["last_date"]["value_as_string"]datetime_obj = datetime.strptime(last_updated_date_string, "%Y-%m-%dT%H:%M:%S.%fZ")return datetime_obj.strftime("%Y-%m-%d")def update_new_data(df, es, last_update_date, index_name):if isinstance(last_update_date, str):last_update_date = datetime.strptime(last_update_date, "%Y-%m-%d")last_update_date = pd.Timestamp(last_update_date).normalize()if not df.empty and "close_approach_date" in df.columns:df["close_approach_date"] = pd.to_datetime(df["close_approach_date"])today = pd.Timestamp(datetime.now().date()).normalize()if df is not None and not df.empty:update_range = df.loc[(df["close_approach_date"] > last_update_date)& (df["close_approach_date"] < today)]print(update_range)if not update_range.empty:helpers.bulk(es, doc_generator(update_range, index_name))else:print("No new data to update.")else:print("The DataFrame is empty or None.")# Triggered from a message on a Cloud Pub/Sub topic.
@functions_framework.cloud_event
def hello_pubsub(cloud_event):index_name = "asteroid_data_set"es = connect_to_elastic()last_update_date = updated_last(es, index_name)print(last_update_date)response = connect_to_nasa(last_update_date)df = create_df(response)try:if df is None:raise ValueError("DataFrame is None. There may be a problem.")update_new_data(df, es, last_update_date, index_name)print(updated_last(es, index_name))except Exception as e:print(f"An error occurred: {e}")

添加 requirements.txt 文件

你还需要定义一个requirements.txt 文件,其中包含运行代码所需的所有指定包。

functions-framework==3.*
requests==2.31.0
elasticsearch==8.12.0
pandas==2.1.4

调度你的云函数

在 Cloud Scheduler 中,你可以将函数设置为使用 unix cron 格式定期运行。 我将代码设置为每天早上 8 点在我的时区运行。

你还需要配置执行以连接到你之前创建的 Pub/Sub 主题。 我目前将消息正文设置为 “hello”。

现在你已经设置了 Pub/Sub 主题和 Cloud Function 并将该 Cloud Function 设置为按计划运行,只要出现新数据,你的索引就会自动更新。

结论

使用 Python、Google Cloud Platform Functions 和 Google Cloud Scheduler,你应该能够确保定期更新索引。 你可以在此处找到完整的代码以及用于本地测试的搜索实验室笔记本。 我们还与 Google Cloud 一起举办了一场点播网络研讨会,如果你想构建搜索应用程序,这可能是一个不错的下一步。 如果你基于此博客构建了任何内容,或者如果你对我们的讨论论坛和社区 Slack 频道有疑问,请告诉我们。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/282542.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

每日OJ题_牛客WY3 小易的升级之路(IO型OJ)

目录 牛客WY3 小易的升级之路 解析代码 牛客WY3 小易的升级之路 小易的升级之路_牛客题霸_牛客网 解析代码 #include <iostream> #include <vector> using namespace std;int getVal(int a, int b) {if (a > b)swap(a, b);for (int i a; i > 1; --i){if…

科研绘图一:箱线图(添加贝赛尔曲线)

R语言绘图系列—箱线图贝赛尔曲线 &#xff08;一&#xff09;: 科研绘图一&#xff1a;箱线图&#xff08;添加贝赛尔曲线&#xff09; 文章目录 R语言绘图系列---箱线图贝赛尔曲线&#xff08;一&#xff09;: 科研绘图一&#xff1a;箱线图&#xff08;添加贝赛尔曲线&…

【MySQL】MySQL用户管理

文章目录 一、用户1.用户信息2.创建用户3.删除用户4.修改用户密码 二、数据库的权限1.给用户授权2.回收权限 一、用户 如果我们只能使用root用户&#xff0c;这样存在安全隐患。这时&#xff0c;就需要使用MySQL的用户管理。 1.用户信息 我们安装mysql之后&#xff0c;会自动…

DC-1靶场

一.环境搭建 下载地址 http://www.five86.com/downloads/DC-1.zip 把桥接设置为nat模式&#xff0c;打开靶机的时候会提示几个错误&#xff0c;点击重试即可 启动靶机&#xff0c;如下图所示即可 二.开始打靶 1.信息收集 arp-scan -l 扫描跟kali&#xff08;攻击机&…

MySQL基础(DDL,DML,DQL)

目录 一DDL 1.1数据库操作 1.1.1查询所有数据库&#xff1a; 1.1.2创建数据库 1.1.3 使用数据库 1.1.4 删除数据库 1.2表操作 1.2.1表操作 1.2.1.1创建表 1.2.1.1.1约束 1.2.1.1.2 数据类型 1.2.1.1.2.1 数值类型 1.2.1.1.2.2 字符串类型 1.2.1.1.2.3日期类型 1.…

Linux:执行命令的命令eval与Bash解析命令的方式

相关阅读 Linuxhttps://blog.csdn.net/weixin_45791458/category_12234591.html?spm1001.2014.3001.5482 eval命令用于接收参数&#xff0c;并将这些参数作为一行命令执行&#xff0c;这也许会使人困惑&#xff0c;为什么我不能直接执行命令而需要使用eval命令间接执行呢&…

【应用笔记】LAT1305+使用STM32+TT类型IO的注意事项

1. 概述 在 STM32 系列 MCU 中&#xff0c; 除了一些特殊管脚外&#xff0c;绝大多数管脚都可以分类为 FT (兼容5V 信号)或 TT&#xff08;兼容 3V3 信号&#xff09;类型的 IO&#xff0c;由于 MCU 内部设计的不同&#xff0c; TT IO 相比 5V IO 有更多的限制&#xff0c;下面…

【回溯专题part1】【蓝桥杯备考训练】:n-皇后问题、木棒、飞机降落【已更新完成】

目录 1、n-皇后问题&#xff08;回溯模板&#xff09; 2、木棒&#xff08;《算法竞赛进阶指南》、UVA307&#xff09; 3、飞机降落&#xff08;第十四届蓝桥杯省赛C B组&#xff09; 1、n-皇后问题&#xff08;回溯模板&#xff09; n皇后问题是指将 n 个皇后放在 nn 的国…

MySQL数据库索引

目录 一.索引的基本内容 1.索引的定义 2.索引的作用 &#xff08;1&#xff09;设置了合适的索引之后&#xff0c;数据库利用各种快速定位技术&#xff0c;能够大大加快查询速度&#xff0c;这是创建索引的最主要的原因。 &#xff08;2&#xff09;当表很大或查询涉及到多…

鸿蒙一次开发,多端部署(四)工程管理

DevEco Studio的基本使用&#xff0c;请参考DevEco Studio使用指南。本章主要介绍如何使用DevEco Studio进行多设备应用开发。 说明&#xff1a; 本章的内容基于DevEco Studio 3.1.1 Release版本进行介绍&#xff0c;如您使用DevEco Studio其它版本&#xff0c;可能存在文档与产…

python --- 练习题3

目录 1、猜数字游戏&#xff08;使用random模块完成&#xff09; &#xff1a;继上期题目&#xff0c;附加 2、用户登录注册案例 3、求50~150之间的质数是那些&#xff1f; 4、打印输出标准水仙花数&#xff0c;输出这些水仙花数 5、验证:任意一个大于9的整数减去它的各位…

行业模板|DataEase制造行业大屏模板推荐

DataEase开源数据可视化分析平台于2022年6月发布模板市场&#xff08;https://templates-de.fit2cloud.com&#xff09;&#xff0c;并于2024年1月新增适用于DataEase v2版本的模板分类。模板市场旨在为DataEase用户提供专业、美观、拿来即用的大屏模板&#xff0c;方便用户根据…

机器学习_正则化

文章目录 代价函数 如果我们有非常多的特征&#xff0c;我们通过学习得到的假设可能能够非常好地适应训练集&#xff08;代价函数可能几乎为 0&#xff09;&#xff0c;但是可能会不能推广到新的数据。 下图是一个回归问题的例子&#xff1a; 第一个模型是一个线性模型&#xf…

思科网络中DHCP协议的配置

一、什么是DHCP&#xff1f;DHCP有什么作用&#xff1f; &#xff08;1&#xff09;DHCP&#xff08;Dynamic Host Configuration Protocol&#xff09;是一种网络协议&#xff0c;用于在局域网中自动分配IP地址和其他网络配置信息给计算机设备。我们电脑连接WiFi的时一般就是…

鸿蒙一次开发,多端部署(三)应用UX设计原则

设计原则 当为多种不同的设备开发应用时&#xff0c;有如下设计原则&#xff1a; 差异性 充分了解所要支持的设备&#xff0c;包括屏幕尺寸、交互方式、使用场景、用户人群等&#xff0c;对设备的特性进行针对性的设计。 一致性 除了要考虑每个设备的特性外&#xff0c;还…

VMD + CEEMDAN 二次分解,CNN-Transformer预测模型

往期精彩内容&#xff1a; 时序预测&#xff1a;LSTM、ARIMA、Holt-Winters、SARIMA模型的分析与比较-CSDN博客 风速预测&#xff08;一&#xff09;数据集介绍和预处理-CSDN博客 风速预测&#xff08;二&#xff09;基于Pytorch的EMD-LSTM模型-CSDN博客 风速预测&#xff…

【Excel自动化办公】使用openpyxl对Excel进行读写操作

目录 一、环境安装 1.1 创建python项目 1.2 安装openpyxl依赖 二、Excel数据读取操作 三、Excel数据写入操作 3.1 创建空白工作簿 3.2 写数据 四、设置单元格样式 4.1 字体样式 4.2 设置单元格背景填充色 4.3 设置单元格边框样式 4.4 单元格对齐方式 4.5 数据筛选…

Manga image Translator: 一键翻译各类图片内文字支持多种语言开源AI工具

项目地址&#xff1a;GitHub - zyddnys/manga-image-translator: Translate manga/image 一键翻译各类图片内文字 https://cotrans.touhou.ai/ 官方演示&#xff1a; https://touhou.ai/imgtrans/ 夸克网盘文件包下载链接&#xff1a;夸克网盘分享提取码&#xff1a;22G3 文…

基于高斯模型的运动目标检测(车辆检测),Matlab实现

博主简介&#xff1a; 专注、专一于Matlab图像处理学习、交流&#xff0c;matlab图像代码代做/项目合作可以联系&#xff08;QQ:3249726188&#xff09; 个人主页&#xff1a;Matlab_ImagePro-CSDN博客 原则&#xff1a;代码均由本人编写完成&#xff0c;非中介&#xff0c;提供…

数值分析(三) Lagrange(拉格朗日)插值法及Matlab代码实现

目录 前言一、Lagrange&#xff08;拉格朗日&#xff09;插值1. 线性插值2. 抛物插值3. 拉格朗日插值多项式 二、Lagrange插值算法及matlab代码1. Lagrange 插值算法matlab实现2 实例3. 线性插值示意图代码4. 抛物插值示意图代码 三、总结四、插值法专栏 前言 本篇为插值法专栏…