后端开发教程 - Java、Spring Boot 实战 - msg200.com
如何首次构建数据管道
步骤 1：获取 API 密钥
第 2 步：保持 API 密钥的私密性
步骤 3：创建一个新的空数据库
步骤 4：检索数据并保存为 json 文件
步骤 5：从 JSON 文件到字典再到 Pandas Dataframe
步骤 6：使用 Pandas 进行 ETL
步骤 7：更新数据库
完毕！
我对机器学习模型及其为世界提供的神奇工具着迷，无论是用于决策、预测趋势、改善生活，还是拯救生命。这些模型基于数据库中不断收集和存储的海量数据进行训练和测试。
我真的很好奇，在训练出出色的机器学习模型之前，原始数据是如何进入程序的，以及要经过哪些过程才能使这些原始数据变得有用。
我们日常用于预测的一种数据是天气观测数据。显然，天气预报主要基于复杂的物理方程和统计模型。不过，我认为建立自己的天气数据库来练习我的数据科学技能应该很有趣。
我没有将温度传感器连接到我的电脑并在家里开设自己的气象站（这对于我的下一个项目来说可能非常酷），而是决定建立一个简单的数据管道来看看它是如何完成的。
我的任务是从头开始构建一个管道 - 从OpenWeatherMap当前天气 Web API 检索天气数据开始，使用 Pandas（python 数据分析）库解析数据并将其存储在本地 SQLite 数据库中。
以下是我的做法：
步骤 1：获取 API 密钥
OpenWeatherMap提供了一些付费 API 方案，其中一些是按月订阅，其中包含功能强大的工具。我很高兴从免费方案开始，因为它可以访问任何城市的最新天气数据。
第 2 步：保持 API 密钥的私密性
你肯定不想让陌生人知道你的 ATM 密码吧？API 密钥也可以用来窃取你的钱包信息。尤其当你是此 API 服务的付费用户，并且 API 调用次数有限的情况下。环境变量在处理这个问题时非常有用，因为它们的值是在程序外部设置的（强烈推荐参考此处的教程）。通过将我的 API 密钥设置为环境变量，它就隐藏在程序外部，并且可以在我的公共 GitHub 存储库中不暴露的情况下使用。我将 API 密钥保存到一个名为 .env 的新文件中。它的内部结构如下：
api-token = "typeyourapikeyhere"
现在我们可以开始编写流水线了。在编写代码之前，我花了一些时间设计我的程序，并最终决定根据职责将程序分成几个文件。
我的程序设计准则如下：
- 每个函数只做一件事。
- 每个函数都单独测试，以确保其真正实现了预期功能。此外，当其他 bug 堆积如山，需要找出哪个函数有问题时，这样做还能省去不少麻烦。这里我选择使用doctest进行一些简单的小测试。
对于后续步骤，请确保：
- Python 3 和 pip 已经安装。
- 如果您使用我的GitHub 存储库中的文件，则可以通过运行以下命令安装依赖项：
pip install -r requirements.txt
- 在运行我的程序的文件夹中，你需要一个名为 data_cache 的文件夹。所有数据在存储到数据库之前都会保存在这里。
步骤 3：创建一个新的空数据库
在 python 文件中，我使用 sqlite3 库创建了一个 SQLite 数据库：
| import sqlite3 | |
| def create_DB_connection(): | |
| """ Connect to a database and create cursor object for executing SQL statements. | |
| """ | |
| connection = sqlite3.connect('CurrentWeatherDB.db', uri=True) | |
| print("Connection with database established ..........") | |
| connection.cursor() | |
| return connection | |
| def close_connection(connection): | |
| """ Close connection with the database. | |
| """ | |
| connection.commit() | |
| connection.close() |
步骤 4：检索数据并保存为 json 文件
此时，您将能够获取 JSON 格式的数据，并将其保存为当前文件夹中的 JSON 文件。每个 JSON 文件都以“dt”值命名，该值代表日期时间 (datetime)。请注意，日期时间格式为 Unix 纪元时间戳 (Unix Epoch Timestamp)。
步骤 5：从 JSON 文件到字典再到 Pandas Dataframe
这听起来可能有点徒劳无功，但我更喜欢把整个过程分解成尽可能多的“小步骤”。对我来说，这样条理清晰，而且有助于追踪每个步骤。
步骤 6：使用 Pandas 进行 ETL
ETL 流程使我们能够根据分析需求提取和转换数据，然后将其加载到数据仓库中。为了使我的数据能够用于未来的数据科学项目，我确保数据库包含每日温度预测所需的参数（当前温度（摄氏度）、最低和最高温度（摄氏度）、湿度、气压和风力）。此外，我选择日期时间（“dt”）列作为数据库主键的行索引。
步骤 7：更新数据库
现在我们已将当前天气数据保存在数据框中，我们可以使用 Pandas 库轻松地将其加载到数据库中。
为了提高灵活性，我使用了 argparse 库，它允许你从命令行运行 main.py，并将城市 ID 作为可选参数传入。因此，即使我默认定义了特拉维夫城市，用户仍然可以针对世界上任何城市运行该程序。例如，如果你想获取美国底特律的天气数据：
python3 src/main.py --city_id "4990729"
运行main.py时，执行步骤3-7：
完毕！
就这样！一个你可以自行构建的管道就完成了。最终，该程序将按计划运行，逐步构建数据库，用于我的下一个数据科学项目。目前，Windows Scheduler是一个很好的入门工具，但我也推荐你尝试一下Python Scheduler。网上有很多精彩的教程，等你来探索。
编辑于 2020 年 10 月 18 日：
- 该程序使用 while 循环按默认计划每 15 分钟运行一次，您可以根据需要设置频率。例如，要每 5 秒运行一次，请运行以下命令：
python3 src/main.py --frequency 5
- 目前，所有 Python 文件都存储在名为 src 的本地文件夹中。在下一篇文章中，我将解释如何创建一个仅复制本项目所需的 Python 文件和依赖项的 Docker 镜像。