上周接了一个内部通知需求，本来只是想把“查询天气”“判断是否要提醒”“发到 Slack”这三件事串起来，结果发现大家嘴上说的 Workflow 其实差别很大。有人想用 Elasticsearch 8.15 之后自带的搜索工作流能力，有人只是在找一个能定时跑脚本的编排器，还有人只是想解决“怎么从 ES 查出结果再扔给 Slack”这种最朴素的场景。这篇文章我不会只贴代码，而是把从需求拆解、环境准备、天气数据入库、查询判定、Slack 通知到最终封装成可调度工作流的完整过程讲清楚。项目标题里那几个词——Elasticsearch、Workflow、Slack——每个都有值得单独展开的坑，我尽量用跑通过的实践来说。
1. 先想清楚：你要编排的到底是什么
1.1 表面需求是“查天气”，真实需求是“有条件的通知”
直接拿一个脚本调天气 API，然后把 temperature 拼成一段文本发到 Slack。这个过程五分钟能写完，但它不算 Workflow，因为它没有“决策”。实际工作中我们要的不是每天早上七点被一条“上海当前温度 12 度”的消息吵醒，而是只在天气达到某个阈值时才收到提醒，比如：
- 当前有中雨以上天气，出门要带伞；
- 风速超过 30 km/h，不适合户外骑行；
- 温度低于 5 度并且天气为雨夹雪，需要提醒值班人员注意路况。
所以这个需求真正困难的地方在于：查询不是目的，在什么条件下把查询结果投递给人才是目的。Workflow 的核心就是把“取数据、存数据、判定、通知”这四个步骤拆成独立环节，同时保留一条清晰的执行链路。
1.2 我当时为什么选择 Elasticsearch 做承载
有人问过：既然天气是从第三方 API 拿到的，为什么还要先写进 Elasticsearch，直接在内存里判断不行吗？单次查询确实可以，但如果接下来要做天气历史趋势分析、按城市维度横向对比、或者把天气数据和业务故障数据放在同一个看板里观察，Elasticsearch 的价值就出来了。Elasticsearch 本身最强的是检索和聚合，不是跑定时调度，所以我的设计思路很朴素：
- 上游：Open-Meteo 或者其他天气 API，负责提供实时天气；
- 中游：Elasticsearch，负责保存时序化天气快照、提供查询和判断依据；
- 下游：Slack Incoming Webhook，负责把筛选后的消息推给对应频道。
整条链路就是一个可复用的工作流。如果你想在 ES 原生能力里找到名为 Workflow 的独立模块，8.15 以后的版本确实在逐步演进搜索工作流，但目前更通用、更可控的方案还是用外部调度脚本把几个动作串起来。先把能稳定跑通的链路做出来，再谈要不要迁移到原生 API。
1.3 数据流和环节职责
下面这张表是我在设计阶段给自己画的，避免写到一半把职责搞混。
| 阶段 | 输入 | 动作 | 输出 |
|---|---|---|---|
| 1. fetch | 城市经纬度 | 调用天气 API | 结构化天气 JSON |
| 2. store | 结构化天气 JSON | 写入 ES weather_metrics 索引 | 可检索的历史记录 |
| 3. query | 城市名 | 查询最新一条并套判定规则 | 是否告警 |
| 4. notify | 告警文本 | POST 到 Slack Webhook | Slack 消息 |
每一步都可以单独替换。比如今天用 Open-Meteo，明天想换和风天气，只需要改 fetch；今天用 Slack，明天想推到企业微信，只需要改 notify。这就是把一个散装脚本整理成 Workflow 的最大收益。
2. 准备一套能跑实验的 Elasticsearch 环境
2.1 Docker Compose 是最省事的方案
本机装过 ES 的人都懂，依赖 JDK、内存配置、证书、权限，哪个环节出问题都能卡半天。我这里直接使用 Docker Compose 起一个单节点集群，集群版本选择 8.13.2。我知道很多人搜过 9.0.4 Windows 版本下载，但实际项目里不必追新，8.x 足够稳，而且下面的配置思路在 9.x 上也通用。
services: es: image: docker.elastic.co/elasticsearch/elasticsearch:8.13.2 container_name: es-weather-demo environment: - discovery.type=single-node - xpack.security.enabled=true - ELASTIC_PASSWORD=your_password - ES_JAVA_OPTS=-Xms2g -Xmx2g ports: - "9200:9200" volumes: - es_data:/usr/share/elasticsearch/data volumes: es_data:启动命令很简单：
docker compose up -d docker compose logs -f es看到"message": "started"一般就是起来了。这里有个新手容易忽略的问题：8.x 默认开启安全认证，并且 Elasticsearch 默认暴露的是 HTTPS 端口，使用浏览器、curl 或者 Python 客户端时，不能再用http://localhost:9200直连。
2.2 在 Windows 上跑 Elasticsearch 的额外注意
如果你没有 Docker 环境，想直接用 Windows 压缩包跑，也不是不行。解压后建议把目录放到不带中文和不带空格的路径下，比如D:\elasticsearch-8.13.2，否则启动时容易因为路径解析出问题。
启动前至少要改两个地方。
一是调整 JVM 堆内存。ES 默认堆内存可能偏小，开发机建议打开config/jvm.options，把-Xms和-Xmx改成相同的值，比如 2g。这里强调相同值是为了避免运行中出现动态扩容导致停顿。
二是确认本机端口没有占用。如果 9200 被占，ES 会启动失败。Windows 下命令：
netstat -ano | findstr :9200没有输出就代表端口空闲。确认没问题后再执行：
bin\elasticsearch.bat不要双击这个 bat。因为一旦直接双击，错误信息一闪而过，你根本不知道是内存问题还是端口问题。在终端里启动至少能看到报错。等窗口稳定后，另开一个 PowerShell 执行：
curl -k -u elastic:your_password https://localhost:9200因为本机是自签名证书，-k用于跳过证书校验。如果看到"tagline" : "You Know, for Search"，ES 就没问题了。
2.3 用 Python 客户端完成连接验证
后面代码会用到 Elasticsearch 官方的 Python 客户端，先安装依赖：
pip install elasticsearch requests然后写一个最简单的健康检查：
from elasticsearch import Elasticsearch es = Elasticsearch( "https://localhost:9200", basic_auth=("elastic", "your_password"), verify_certs=False, request_timeout=30 ) print(es.info())注意这里连接串写的是 HTTPS。如果只想在本地开发验证而不想处理证书，可以使用verify_certs=False，但生产环境务必用 CA 证书或者 API Key，不要整体关闭校验。
3. 上游数据：用 Open-Meteo 把天气拉下来写进 ES
3.1 为什么不选那些看起来很复杂的天气服务
天气 API 有很多选择，但大家会发现很多服务要先注册、要领 API Key、还限制调用次数。Open-Meteo 是一个很适合做实验的免费天气接口，不需要 Key，直接通过 HTTP GET 就能拿到预报数据。它的 URL 大概长这样：
https://api.open-meteo.com/v1/forecast?latitude=31.2304&longitude=121.4737¤t_weather=truelatitude 是纬度，longitude 是经度，current_weather=true表示只返回当前天气而不是未来小时级预报。返回的 JSON 是这样：
{ "latitude": 31.23, "longitude": 121.47, "current_weather": { "temperature": 12.4, "windspeed": 18.3, "winddirection": 120, "weathercode": 3, "time": "2025-06-20T10:15" } }实际生产场景中要想清楚：调用免费 API 是否有稳定性和数据授权问题，但在技术 demo 和内部原型阶段，它够用了。
3.2 把天气快照写入 Elasticsearch 的正确姿势
天气数据本质上是一种小体量时序数据。我建议不要裸调 API 后直接用结果发 Slack，而是先写入索引，这样后续想查看历史、想用 Kibana 画趋势图都有数据可用。
先创建索引映射。天气字段里面存在不少数字，如果不提前声明映射，ES 会根据第一条数据自动推断。自动映射在 demo 里问题不大，但我更建议手动创建映射，因为后面查询和聚合会更可控。在 Kibana Dev Tools 中执行：
PUT /weather_metrics { "mappings": { "properties": { "@timestamp": { "type": "date" }, "city": { "type": "keyword" }, "temperature": { "type": "float" }, "windspeed": { "type": "float" }, "winddirection": { "type": "short" }, "weathercode": { "type": "keyword" }, "record_time": { "type": "keyword" } } } }需要注意，record_time 是天气接口返回的当地观测时间，我把它设计成 keyword，而不是 date。原因是 Open-Meteo 返回的时间字符串没有带时区，如果强行映射为 date，ES 会按 UTC 解析，和国内本地时间直观对比时容易差 8 个小时。排序和筛选统一使用@timestamp，这个字段代表数据写入 Elasticsearch 的 UTC 时间，语义清晰。
Python 侧拉取并写入：
from datetime import datetime, timezone import requests from elasticsearch import Elasticsearch es = Elasticsearch( "https://localhost:9200", basic_auth=("elastic", "your_password"), verify_certs=False, request_timeout=30 ) def fetch_weather(city_name: str, lat: float, lon: float) -> dict: resp = requests.get( "https://api.open-meteo.com/v1/forecast", params={ "latitude": lat, "longitude": lon, "current_weather": "true" }, timeout=15 ) resp.raise_for_status() cur = resp.json()["current_weather"] return { "@timestamp": datetime.now(timezone.utc).isoformat(), "city": city_name, "temperature": cur["temperature"], "windspeed": cur["windspeed"], "winddirection": cur.get("winddirection"), "weathercode": str(cur["weathercode"]), "record_time": cur["time"] } def store_weather(doc: dict) -> str: doc_id = f"{doc['city']}-{doc['record_time']}" return es.index( index="weather_metrics", id=doc_id, document=doc )["_id"]这里值得展开一下 doc_id 的设计。如果我们每天定时跑任务，Open-Meteo 在某个观测时刻返回的天气数据是同一个天气事件。使用城市-观测时间作为 document id，可以在同一时刻重复跑任务时覆盖同一条记录，而不是反复新增。批量采集多个城市时，也不需要额外装什么 bulk 插件，官方客户端自带批量处理能力，数据量大再改成 elasticsearch.helpers.bulk 就行。
3.3 先落库存查询到底多了哪些好处
很多人觉得多了一步写 ES 很麻烦，我实际用下来觉得有三个价值。
第一，历史数据可追溯。如果某天 Slack 推了一条“暴雨预警”，事后复盘时可以直接在 ES 里查当时的天气数据，确认告警是否合理。
第二，判断逻辑和采集逻辑解耦。天气采集可能需要每十分钟一次，但通知只需要每天一次或者达到阈值才触发。如果每次采集后都直接执行通知逻辑，后面做降噪会非常痛苦。把数据先存下来，通知环节按需查询，代码会清爽很多。
第三，可以和已有业务数据联动。比如把天气数据和当天的订单量、故障工单量放到同一张可视化看板，观察天气对业务指标的影响。这在纯 API 脚本方案里很难做，但数据一旦进入 ES，就能用现成的聚合能力去分析。
4. 让查询决定要不要发：ES 查询和天气告警逻辑
4.1 怎么查出某城市最新一条天气记录
ES 查询要返回“最新一条”，不能直接 size=1 然后碰运气，必须按时间排序。查询语句如下：
GET /weather_metrics/_search { "size": 1, "query": { "bool": { "filter": [ { "term": { "city": "Shanghai" } } ] } }, "sort": [ { "@timestamp": { "order": "desc" } } ] }Python 侧封装成函数后，就能在后续 Workflow 中复用了。
def get_latest_weather(city: str) -> dict | None: resp = es.search( index="weather_metrics", size=1, query={"term": {"city": city}}, sort=[{"@timestamp": {"order": "desc"}}], source=["city", "temperature", "windspeed", "weathercode", "record_time"] ) hits = resp["hits"]["hits"] if not hits: return None return hits[0]["_source"]这里有个小优化：通过source只请求需要的字段，避免把整个文档完整拉回来。数据量小时不明显，但数据量上来了之后能减少网络传输和内存压力。
4.2 把天气编码翻译成“要不要提醒”
Open-Meteo 的 weathercode 是一组数字编码，想要理解这些编码，我对着文档归纳了一张简表。这张表不需要特别精确，因为最终判定规则可以根据业务调整。
| weathercode | 含义 |
|---|---|
| 0 | 晴 |
| 1-2 | 少云到多云 |
| 3 | 阴 |
| 45/48 | 雾 |
| 51-57 | 毛毛雨 |
| 61-67 | 雨 |
| 71-77 | 雪 |
| 80-82 | 阵雨 |
| 95-99 | 雷暴 |
我当时的告警规则是：雨相关编码命中 61、63、65、80、81、82 时提醒带伞，命中 95 到 99 时因为可能伴随雷电，优先级提高；另外单独把风速大于等于 30 km/h 当作另一个独立条件。这个 30 不是公式算出来的，是结合户外活动经验定的，不同团队完全应该按自己的场景改。Python 判定函数可以这样写：
RAINY_CODES = {"61", "63", "65", "80", "81", "82"} def should_alert(source: dict) -> bool: code = source.get("weathercode", "") wind = float(source.get("windspeed", 0) or 0) return code in RAINY_CODES or wind >= 304.3 组装一段能看懂的 Slack 文本
判定通过后不能直接把 JSON 原封不动发到 Slack，那样群里全是代码噪音。最好整理成人类可读的文本。我当时的做法是做一层简单的编码翻译，再做一次字符串拼接。
CODE_LABEL = { "0": "晴天", "1": "少云", "3": "阴天", "61": "小雨", "63": "中雨", "65": "大雨", "80": "阵雨", "95": "雷阵雨" } def build_message(source: dict) -> str: city = source["city"] code = source.get("weathercode", "unknown") label = CODE_LABEL.get(code, f"未知天气码({code})") temp = source["temperature"] wind = source["windspeed"] record_time = source.get("record_time", "") reason = [] if code in RAINY_CODES: reason.append("有降雨") if float(wind) >= 30: reason.append("风速偏大") return ( f"{city} 天气提醒\n" f"观测时间：{record_time}\n" f"天气：{label}\n" f"温度：{temp}℃\n" f"风速：{wind} km/h\n" f"提醒原因：{'，'.join(reason)}" )我在实际调试中发现一个坑：Open-Meteo 返回的 weathercode 是数字，比如 61。如果索引映射里定义为 keyword，Python 侧也方便比较，但如果你从 JSON 里直接取数后发现是 int，而你的判定集合里写的是字符串，两者永远不相等。建议在 fetch 阶段就做str(cur["weathercode"])统一转字符串。数据来源多时，类型不统一是告警逻辑失效的重灾区。
5. 发 Slack 通知：连接 Webhook 这一步没想象中简单
5.1 创建 Slack Incoming Webhook
Slack 发送消息最快速的方式是 Incoming Webhook。打开 Slack 的管理后台，进入 API 页面，创建一个 App，然后在 Incoming Webhooks 开关下添加一个 Webhook，选好目标频道，Slack 会给你生成一个形如下面的长地址：
https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX这里的 Webhook 相当于一个带写入权限的 URL，谁拿到这个地址，谁就能往对应频道发消息。因此不要把 Webhook 直接写死在仓库里。我会建议用一个.env文件保存，然后在 Python 中读取：
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/xxxx5.2 用 Python 发送一条消息
Slack 的 Incoming Webhook 要求 POST 一个 JSON，核心字段就是text。最简单的代码：
import os def send_slack(text: str) -> bool: webhook_url = os.environ.get("SLACK_WEBHOOK_URL") if not webhook_url: print("SLACK_WEBHOOK_URL 未配置，发送失败") return False resp = requests.post( webhook_url, json={"text": text}, timeout=15 ) if resp.status_code != 200: print(f"Slack 返回异常：{resp.status_code} {resp.text}") return False return True正常返回是 HTTP 200 并且响应体为ok。很多教程只贴一个 curl 命令，但发消息之前需要留意两点。
第一，Slack 消息内容里的换行符必须是字符串里的真实\n，如果从某个配置系统里读多行文本，注意别把\\n原样传过去。
第二，如果消息内容后续要放入 JSON，尽量使用json={"text": text}而不是手动拼接 JSON 字符串。手动拼接时一旦文本里出现引号，整个 payload 就会格式错误。
5.3 消息发送的“成功”要如何定义
我之前犯过的一个错误是只看 HTTP 200 就认为发送成功。Incoming Webhook 基本可以这样做，但严格一点还要确认响应体里没有 error。响应体正常是一个单词ok，如果出现类似invalid_payload的报错，说明提交的 JSON 结构不对。如果遇到 HTTP 429，说明触发了 Slack 的限流，需要退避重试。
if resp.status_code == 429: print("Slack 限流，建议稍后重试")Slack 本身是外部服务，网络超时和限流都可能发生。在 Workflow 中加入简单的超时和重试逻辑，能有效减少“脚本没报错但消息没发出”的悬案。
6. 把整个流程封装成一个可调度的 Workflow
6.1 先把各种函数整理成一个类
到这一步，我们已经有了 fetch_weather、store_weather、get_latest_weather、should_alert、build_message、send_slack 这些函数。如果继续在脚本里从上到下顺序调用，并非不能用，但后续要加城市循环、加条件判断会越来越乱。
我建议封装成一个 WeatherWorkflow 类，把四个核心步骤暴露成有名字的方法：
class WeatherWorkflow: def __init__(self, es_client, slack_webhook_url): self.es = es_client self.slack_webhook_url = slack_webhook_url def run(self, city: str, lat: float, lon: float) -> None: doc = self._fetch_weather(city, lat, lon) self._store_weather(doc) latest = self._get_latest_weather(city) if not latest: print("没有查到最新天气记录") return if not should_alert(latest): print(f"{city} 当前天气不满足提醒条件，跳过 Slack 发送") return text = build_message(latest) self._send_slack(text) def _fetch_weather(self, city, lat, lon): return fetch_weather(city, lat, lon)这里的优点不是代码变少，而是当执行顺序需要变化时，你只要改 run 方法的行序。有人可能觉得这个类叫 Workflow 有点“重”，但把整条业务链路的入口收敛成一个 run() 方法，远比在多个脚本之间复制粘贴更便于维护。
6.2 调度方式：每天定时执行
让 workflow 跑起来需要一个触发机制。Linux 下用 cron 就可以，写一个工作目录下的执行脚本：
0 7 * * * cd /opt/weather_job && python -m workflow.main如果想在 Windows 上做定时，可以用任务计划程序。我简单说下命令行方式，这也是很多人搜“windows 启动 elasticsearch”之后会遇到的任务配置问题：
schtasks /create /tn "weather_slack" /tr "python C:\weather_job\main.py" /sc daily /st 07:00不过在 Windows 上直接写python有可能找不到解释器，因为系统没有把 Python 路径加入 PATH。先执行where python查一下完整路径，再放进任务计划里比较稳妥。这一类问题看起来小，实际排查起来很费时间。
6.3 避免重复告警：加一个状态标记
定时任务跑起来以后，一定会遇到重复告警问题。比如当天 7 点调度了一次，天气正好是中雨，发送了一条 Slack；如果 8 点又调度了一次，天气还是中雨，理论上这条消息不需要再次发送，否则群里会变得很吵。
我推荐的方案是在 ES 里增加一个 weather_alert_state 索引，记录每个城市已发送过的天气事件标识。发送前先查询是否已经发过。标识可以用城市-天气观测时间-天气编码生成。
def _already_notified(city: str, record_time: str, code: str) -> bool: marker = f"{city}-{record_time}-{code}" return es.exists(index="weather_alert_state", id=marker) def _mark_notified(city: str, record_time: str, code: str) -> None: marker = f"{city}-{record_time}-{code}" es.index( index="weather_alert_state", id=marker, document={"marker": marker, "notified_at": datetime.now(timezone.utc).isoformat()} )发送成功后调用_mark_notified，发送前调用_already_notified。这样相同天气事件重复触发时，不会每条都推到 Slack。这里有一个取舍：如果先把“未通知”标记为“已通知”再去调 Slack，万一 Slack 发送失败，这条提醒就永久丢失了；所以我实际是发送成功后再写状态标记，极端情况下可能因为网络超时造成重复发送一次，但比漏掉重要提醒更容易接受。
6.4 单独聊一句：ES 原生 Workflow API 的定位
项目标题里出现 Workflow 时，有人期待的是 ES 原生 Workflow API。在 8.15 之后的版本里，Elasticsearch 确实在搜索侧引入了工作流能力，用来把查询、推理、后处理这些动作编排到一个请求里，典型场景是搜索增强、向量召回和 LLM 结合。但它解决的是“一次搜索请求内多个 Elasticsearch 动作的编排”，不是
