数据工程基础：实践指南

构建数据工程 ETL 管道的实用指南。本指南提供了一种理解和实施数据工程基础知识的实践方法，涵盖存储、处理、自动化和监控。

什么是数据工程？

数据工程专注于组织、处理和自动化数据工作流程，将原始数据转化为有价值的见解，以供分析和决策。 本指南涵盖：

1. 数据存储：定义数据的存储位置和方式。
2. 数据处理：清理和转换原始数据的技术。
3. 工作流程自动化：实现无缝高效的工作流程执行。
4. 系统监控：确保整个数据管道的可靠性和平稳运行。

让我们探索每个阶段！

设置您的开发环境

在我们开始之前，请确保您具备以下条件：

1. 环境设置：
   • 基于 Unix 的系统 (macOS) 或适用于 Linux 的 Windows 子系统 (WSL)。
   • 已安装 Python 3.11（或更高版本）。
   • PostgreSQL 数据库在本地安装并运行。
2. 先决条件：
   • 基本的命令行熟练程度。
   • 基础 Python 编程知识。
   • 软件安装和配置的管理权限。
3. 架构概述：

该图说明了管道组件之间的交互。这种模块化设计充分利用了每种工具的优势：用于工作流程编排的 Airflow、用于分布式数据处理的 Spark 以及用于结构化数据存储的 PostgreSQL。

1. 安装必要的工具：
   • PostgreSQL：

     <code class="language-bash">brew update
     brew install postgresql</code>

     登录后复制

     登录后复制

     登录后复制
   • PySpark：

     <code class="language-bash">brew install apache-spark</code>

     登录后复制

     登录后复制

     登录后复制
   • 气流：

     <code class="language-bash">python -m venv airflow_env
     source airflow_env/bin/activate  # macOS/Linux
     pip install "apache-airflow[postgres]==" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.4/constraints-3.11.txt"
     airflow db migrate</code>

     登录后复制

     登录后复制

     登录后复制

环境准备好了，我们来深入研究各个组件。

1.数据存储：数据库和文件系统

数据存储是任何数据工程管道的基础。 我们将考虑两个主要类别：

• 数据库：有效组织的数据存储，具有搜索、复制和索引等功能。示例包括：
  • SQL 数据库： 用于结构化数据（例如 PostgreSQL、MySQL）。
  • NoSQL 数据库： 用于无模式数据（例如 MongoDB、Redis）。
• 文件系统：适合非结构化数据，提供的功能比数据库少。

设置 PostgreSQL

1. 启动 PostgreSQL 服务：

<code class="language-bash">brew update
brew install postgresql</code>

2. 创建数据库、连接并创建表：

<code class="language-bash">brew install apache-spark</code>

3. 插入示例数据：

<code class="language-bash">python -m venv airflow_env
source airflow_env/bin/activate  # macOS/Linux
pip install "apache-airflow[postgres]==" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.4/constraints-3.11.txt"
airflow db migrate</code>

您的数据现已安全地存储在 PostgreSQL 中。

2.数据处理：PySpark 和分布式计算

数据处理框架将原始数据转化为可操作的见解。 Apache Spark 以其分布式计算能力成为热门选择。

• 处理模式：
  • 批处理：以固定大小的批次处理数据。
  • 流处理：实时处理数据。
• 常用工具： Apache Spark、Flink、Kafka、Hive。

使用 PySpark 处理数据

1. 安装 Java 和 PySpark：

<code class="language-bash">brew services start postgresql</code>

2. 从 CSV 文件加载数据：

使用以下数据创建 sales.csv 文件：

<code class="language-sql">CREATE DATABASE sales_data;
\c sales_data
CREATE TABLE sales (
    id SERIAL PRIMARY KEY,
    item_name TEXT,
    amount NUMERIC,
    sale_date DATE
);</code>

使用以下Python脚本加载和处理数据：

<code class="language-sql">INSERT INTO sales (item_name, amount, sale_date)
VALUES ('Laptop', 1200, '2024-01-10'),
       ('Phone', 800, '2024-01-12');</code>

3. 过滤高价值销售：

<code class="language-bash">brew install openjdk@11 && brew install apache-spark</code>

4. 设置 Postgres DB 驱动程序： 如果需要，请下载 PostgreSQL JDBC 驱动程序并更新下面脚本中的路径。

5. 将处理后的数据保存到 PostgreSQL：

<code class="language-bash">brew update
brew install postgresql</code>

Spark数据处理完成。

3.工作流程自动化：气流

自动化使用调度和依赖关系定义简化工作流程管理。 Airflow、Oozie 和 Luigi 等工具有助于实现这一点。

使用 Airflow 自动化 ETL

1. 初始化气流：

<code class="language-bash">brew install apache-spark</code>

2. 创建工作流 (DAG)：

<code class="language-bash">python -m venv airflow_env
source airflow_env/bin/activate  # macOS/Linux
pip install "apache-airflow[postgres]==" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.4/constraints-3.11.txt"
airflow db migrate</code>

此 DAG 每天运行，执行 PySpark 脚本，并包含验证步骤。 失败时会发送电子邮件警报。

3. 监控工作流程：将 DAG 文件放入 Airflow 的 dags/ 目录中，重新启动 Airflow 服务，并通过 http://localhost:8080 处的 Airflow UI 进行监控。

4.系统监控

监控确保管道可靠性。 Airflow 的警报或与 Grafana 和 Prometheus 等工具的集成是有效的监控策略。 使用 Airflow UI 检查任务状态和日志。

结论

您已经学会了设置数据存储、使用 PySpark 处理数据、使用 Airflow 自动化工作流程以及监控系统。 数据工程是一个关键领域，本指南为进一步探索奠定了坚实的基础。 请记住查阅提供的参考资料以获取更深入的信息。

以上是数据工程基础：实践指南的详细内容。更多信息请关注PHP中文网其他相关文章！