基于Kafka、Spark、Airflow、PostgreSQL和Docker的端到端数据工程实战教程

这篇文章是属于一个分成两个主要阶段的项目的一部分。第一阶段侧重于构建一个数据流水线，这包括从API获取数据，并将这些数据存储到PostgreSQL数据库中。到了第二阶段，我们会开发一个应用，这个应用会用语言模型来操作数据库。

非常适合数据系统或语言模型的应用的新手，这个项目分为两个段落：这两个段落。

• 本文介绍了如何使用Kafka进行流处理、使用Airflow编排工作流、使用Spark进行数据转换以及使用PostgreSQL存储数据，来构建数据管道。我们将利用Docker来设置和运行这些工具。
• 第二篇文章将稍后发布，将介绍如何使用诸如LangChain之类的工具来创建与外部数据库通信的代理程序。

这个项目的第一部分非常适合数据工程初学者，同时也适合希望深入了解整个数据处理生命周期的数据科学家和机器学习工程师。亲自使用这些工具非常有益，有助于改进机器学习模型的设计与扩展，确保它们在实际应用中表现出色。

这篇文章更注重实用而非理论。想深入了解这些工具的内部运作，网上有很多不错的资源可以参考。

概要

让我们一步一步地分解数据管道的流程：

1. 数据流式传输：最初，数据从API流式传输到Kafka主题。
2. 数据处理：接着，Spark作业接手，从Kafka主题消费数据，并将其转移至PostgreSQL数据库。
3. 通过Airflow调度：流处理任务和Spark作业都通过Airflow进行编排。实际上，Kafka生产者会持续监听API，但为了演示，我们将Kafka流处理任务每天运行一次。流处理完成后，Spark作业开始处理数据，使其可以被LLM应用利用。

所有这些工具都将使用 Docker 来构建和运行，更具体地说，是使用 docker-compose 这个工具。

数据管道的概览。图片由作者提供。

有了这个设计图之后，让我们来看看具体的技术细节吧！

首先，你可以使用以下命令在本地机器上克隆 Github 仓库到本地机器：

使用以下命令：git clone https://github.com/example/repo

在终端中运行以下命令克隆代码库:

git clone https://github.com/HamzaG737/data-engineering-project.git

以下是项目的大致结构：

    ├── LICENSE  
    ├── README.md  
    ├── airflow  
    │   ├── Dockerfile  
    │   ├── __init__.py  
    │   └── dags  
    │       ├── __init__.py  
    │       └── dag_kafka_spark.py  
    ├── data  
    │   └── last_processed.json  
    ├── docker-compose-airflow.yaml  
    ├── docker-compose.yml  
    ├── kafka  
    ├── requirements.txt  
    ├── spark  
    │   └── Dockerfile  
    └── src  
        ├── __init__.py  
        ├── constants.py  
        ├── kafka_client  
        │   ├── __init__.py  
        │   └── kafka_stream_data.py  
        └── spark_pgsql  
            └── spark_streaming.py

• airflow 目录包含一个自定义的 Dockerfile 用于设置 airflow，并且包含一个用于创建和调度任务的 [dags](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html) 目录。

• data 目录包含一个 _lastprocessed.json 文件，该文件对于 Kafka 流处理任务至关重要。此文件的角色将在 Kafka 部分详细介绍。

• docker-compose-airflow.yaml 文件定义了运行 Airflow 所需的所有服务。

• docker-compose.yaml 文件指定了 Kafka 服务，并包含一个 docker-proxy。此代理对于通过 Airflow 的 docker-operator 执行 Spark 作业至关重要。此概念将在后续部分详细说明。

• spark 目录包含一个用于设置 Spark 的自定义 Dockerfile。

• src 目录包含运行应用程序所需的所有 Python 模块。

要设置你的本地开发环境，首先安装所需的 Python 包。唯一必需的包是 psycopg2-binary。你可以只安装这个包，或者安装 requirements.txt 文件中列出的所有其他包。要安装所有包，在命令行中运行如下命令。

在命令行中输入以下命令来安装所需的包：`pip install -r requirements.txt`

接下来我们一步步了解项目细节。

该API名为RappelConso，来自法国公共服务。它提供了法国专业人员申报的产品召回数据的访问。这些数据用法文呈现，最初包含31个字段（或列）。其中一些最重要的字段比如：

• _referencefiche (参考文件): 它将在我们的Postgres数据库中作为主键。
• _categorie_deproduit (产品类别): 例如食品、电器、工具、交通工具等。
• _sous_categorie_deproduit (产品子类别): 例如，食品类别的子类别可以是肉类、乳制品、谷物等。
• _motif_derappel (召回原因): 这是最重要的字段之一。
• _date_depublication 即发布日期。
• _risques_encourus_par_leconsommateur 包含消费者在使用该产品时可能遇到的风险。
• 还有一些字段对应不同的链接，例如，产品图片链接、分销商列表链接等。

你可以通过这个链接查看一些示例，并可以通过该链接手动查询数据。

我们在几个关键的数据字段上进行了调整：

1. 之前作为版本控制系统一部分的列如 ndeg_de_version 和 rappelguid 已被移除，因为它们对于我们项目来说并不需要。
2. 我们将涉及消费者风险的列 — risques_encourus_par_le_consommateur 和 description_complementaire_du_risque — 合并，以便更清晰地了解产品风险情况。
3. 表示营销期间的列 date_debut_fin_de_commercialisation 已被拆分成两个独立的列。这样可以更方便地查询产品的营销开始和结束日期。
4. 我们移除了所有列中的重音符号，除了链接、参考编号和日期。这是因为一些文本处理工具处理带重音符号的字符时会遇到困难。

要详细了解这些更改，请查看我们的转换脚本文件 src/kafka_client/transformations.py。请在 src/constants.py 文件中的 DB_FIELDS 查看更新后的列信息。

为了防止每次运行流任务时都从API获取所有数据，我们定义了一个本地的json文件，其中包含最新流的最后更新日期。然后我们就会用这个日期作为新流任务的起始时间。

举个例子，假设最新召回的产品的发布日期是 2023年11月22日。如果我们假设在11月22日之前的所有召回产品信息都已经保存在我们的Postgres数据库里，我们现在可以从11月22日开始逐步传输数据。可能存在重叠，因为可能还有11月22日未处理的数据。

文件保存在 ./data/last_processed.json，其格式为：

    {上次处理时间:"2023-11-22"}

默认，文件是一个空的 json，这意味着我们的首个流处理作业将处理1万条API记录。

注意，这种方法在生产环境中并不可行，更适合使用外部数据库或对象存储。

该用于Kafka流处理的代码可以在./src/kafka_client/kafka_stream_data.py找到，主要涉及从API获取数据，进行数据转换，去除潜在重复数据，更新最后发布日期，并通过Kafka生产者发布数据。

下一步是运行如下定义的 Docker Compose 中的 Kafka 服务：

    version: '3'  

    services:  
      kafka:  
        image: 'bitnami/kafka:latest'  
        ports:  
          - '9094:9094'  
        networks:  
          - airflow-kafka  
        environment:  
          - KAFKA_CFG_NODE_ID=0  
          - KAFKA_CFG_PROCESS_ROLES=controller,broker  
          - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094  
          - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,EXTERNAL://localhost:9094  
          - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT  
          - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093  
          - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER  
        volumes:  
          - ./kafka:/bitnami/kafka  

      kafka-ui:  
        container_name: kafka-ui-1  
        image: provectuslabs/kafka-ui:latest  
        ports:  
          - 8800:8080    
        depends_on:  
          - kafka  
        environment:  
          KAFKA_CLUSTERS_0_NAME: local  
          KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: PLAINTEXT://kafka:9092  
          DYNAMIC_CONFIG_ENABLED: 'true'  
        networks:  
          - airflow-kafka  

    networks:  
      airflow-kafka:  
        external: true

这份文件的主要内容包括如下：

• Kafka 服务使用基础镜像 bitnami/kafka。
• 我们配置服务时只使用了一个 broker，这对我们的小项目来说已经足够了。一个 Kafka broker 负责从生产者（即数据源）接收消息，存储这些消息，并将它们传递给消费者（即数据的最终用户）。broker 监听内部集群通信的端口 9092 和外部通信的端口 9094，允许 Docker 网络外的客户端连接到 Kafka broker。
• 在 volumes 部分，我们将本地目录 kafka 映射到 Docker 容器目录 /_bitnami/kafka_，以确保数据持久化并方便我们从主机系统访问 Kafka 的数据。
• 我们设置了一个名为 kafka-ui 的服务，使用了 Docker 镜像 provectuslabs/kafka-ui:latest。这为与 Kafka 集群交互提供了用户界面，特别适合用来监控和管理 Kafka 主题和消息。
• 为了确保 kafka 和外部分开运行的 airflow 之间的通信，我们将使用外部网络 airflow-kafka 以实现它们之间的通信。

在运行 kafka 服务之前，让我们用以下命令创建 Airflow-Kafka 网络：

    docker network create airflow-kafka

运行上述命令来创建一个名为 airflow-kafka 的 Docker 网络。

现在都准备好了，终于可以开始运行我们的Kafka服务了

运行 `docker-compose up` （启动 Docker 容器）

服务启动之后，访问[kafka-ui]：http://localhost:8800/。你应该会看到类似下面的内容：

Kafka UI界面概览。图片由作者提供。

接下来我们将创建一个包含API消息的主题（topic）。点击左侧的Topics，然后点击左上角的添加主题。主题名称为rappel_conso，由于我们只有一个代理，我们将复制因子（副本因子）设置为1。我们还将分区数设置为1，因为我们一次只有一个消费者线程，因此不需要并行处理。最后，我们将数据保留时间设置为较短，比如一小时，因为我们将在运行Kafka流任务后立即运行Spark作业，所以不需要长时间在Kafka主题中保留数据。

在设置我们的Spark和Airflow配置之前，让我们先创建一个用于持久化我们API数据的Postgres数据库。我使用了PgAdmin 4工具来完成这项任务，不过任何其他的Postgres开发平台也同样可以胜任这项任务。

要安装 PostgreSQL 和 pgadmin，请访问此链接 https://www.postgresql.org/download/ 并下载适用于您操作系统的安装包。在安装 PostgreSQL 的时候之后，您需要设置一个密码，我们之后会用到这个密码从 Spark 环境连接到这个数据库。您也可以让端口保持为默认值 5432。

如果你的安装成功了，你可以打开pgadmin，你应该能看到类似这样的窗口：

pgAdmin界面简介。

因为我们有很多列，我们选择用脚本创建表并添加列，该脚本使用psycopg2，一个Python的PostgreSQL数据库适配器。

你可以用以下命令运行脚本：

你可以用命令 python script.py 来运行脚本。

python scripts/create_table.py

注意，先在脚本里将 postgres 密码设置为环境变量，并命名为 _POSTGRESPASSWORD 。因此，如果你用其他方法得到密码，需要根据实际情况修改脚本。

设置好我们的Postgres数据库后，我们来详细看看这个Spark作业。目标是从Kafka主题_rappel_conso_流式传输数据到Postgres表_rappel_consotable。

     from pyspark.sql import SparkSession  
    from pyspark.sql.types import (  
        StructType,  
        StructField,  
        StringType,  
    )  
    from pyspark.sql.functions import from_json, col  
    from src.constants import POSTGRES_URL, POSTGRES_PROPERTIES, DB_FIELDS  
    import logging  

    logging.basicConfig(  
        level=logging.INFO, format="%(asctime)s:%(funcName)s:%(levelname)s:%(message)s"  
    )  

    def create_spark_session() -> SparkSession:  
        spark = (  
            SparkSession.builder.appName("PostgreSQL Connection with PySpark")  
            .config(  
                "spark.jars.packages",  
                "org.postgresql:postgresql:42.5.4,org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0",  

            )  
            .getOrCreate()  
        )  

        logging.info("Spark session created successfully")  
        return spark  

    def create_initial_dataframe(spark_session):  
        """  
        读取流式数据并创建初始数据框。  
        """  
        try:  
            # 从主题 'rappel_conso' 获取流式数据  
            df = (  
                spark_session.readStream.format("kafka")  
                .option("kafka.bootstrap.servers", "kafka:9092")  
                .option("subscribe", "rappel_conso")  
                .option("startingOffsets", "earliest")  
                .load()  
            )  
            logging.info("Initial dataframe created successfully")  
        except Exception as e:  
            logging.warning(f"Initial dataframe couldn't be created due to exception: {e}")  
            raise  

        return df  

    def create_final_dataframe(df):  
        """  
        修改初始数据框，并创建最终数据框。  
        """  
        schema = StructType(  
            [StructField(field_name, StringType(), True) for field_name in DB_FIELDS]  
        )  
        df_out = (  
            df.selectExpr("CAST(value AS STRING)")  
            .select(from_json(col("value"), schema).alias("data"))  
            .select("data.*")  
        )  
        return df_out  

    def start_streaming(df_parsed, spark):  
        """  
        开始流式传递到 postgres 表 spark_streaming.rappel_conso  
        """  
        # 从 PostgreSQL 读取现有数据  
        existing_data_df = spark.read.jdbc(  
            POSTGRES_URL, "rappel_conso", properties=POSTGRES_PROPERTIES  
        )  

        unique_column = "reference_fiche"  

        logging.info("Start streaming ...")  
        query = df_parsed.writeStream.foreachBatch(  
            lambda batch_df, _: (  
                batch_df.join(  
                    existing_data_df, batch_df[unique_column] == existing_data_df[unique_column], "leftanti"  
                )  
                .write.jdbc(  
                    POSTGRES_URL, "rappel_conso", "append", properties=POSTGRES_PROPERTIES  
                )  
            )  
        ).trigger(once=True) \  
            .start()  

        return query.awaitTermination()  

    def write_to_postgres():  
        spark = create_spark_session()  
        df = create_initial_dataframe(spark)  
        df_final = create_final_dataframe(df)  
        start_streaming(df_final, spark=spark)  

    if __name__ == "__main__":  
        write_to_postgres()

让我们来分析一下这个Spark任务的关键特点和功能：

首先，我们来创建Spark会话环境

    def 初始化Spark会话() -> SparkSession:  
        spark = (  
            SparkSession.builder应用名称("PostgreSQL 连接与 PySpark")  
            .配置(  
                "spark.jars.packages",  
                "org.postgresql:postgresql:42.5.4,org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0",  
            )  
            .获取或创建()  
        )  

        记录.info("Spark 会话启动成功")  
        return spark

2. create_initial_dataframe函数利用Spark的结构化流处理功能从Kafka主题接收实时数据。

    def 创建初始数据集(spark_session):  
        """  
        读取流数据并创建初始数据集。  
        """  
        try:  
            # 从主题 random_names 获取流数据  
            df = (  
                spark_session.readStream.format("kafka")  
                .option("kafka.bootstrap.servers", "kafka:9092")  
                .option("subscribe", "rappel_conso")  
                .option("startingOffsets", "earliest")  
                .load()  
            )  
            logging.info("成功创建了初始数据集")  
        except Exception as e:  
            logging.warning(f"无法创建初始数据集，原因: {e}")  
            raise  

        return df

3. 一旦数据被摄取后，create_final_dataframe 会对其进行转换。它将传入的 JSON 数据应用一个由 DB_FIELDS 定义的模式，使数据结构化，并准备好进一步处理。

    def create_final_dataframe(df):  
        """  
        修改初始数据框并生成最终数据框。  
        """  
        schema = StructType(  
            [StructField(field_name, StringType(), True) for field_name in DB_FIELDS]  
        )  
        df_out = (  
            df.selectExpr("CAST(value AS STRING)")  
            .select(from_json(col("value"), schema).alias("data"))  
            .select("data.*")  
        )  
        return df_out

4. start_streaming 函数读取数据库中的现有数据，将现有数据与传入的数据流进行比较，并添加新的记录。

    def start_streaming(df_parsed, spark):  
        """  
        启动数据流到 spark_streaming.rappel_conso 表  
        """  
        # 从 PostgreSQL 读取现有数据  
        existing_data_df = spark.read.jdbc(  
            POSTGRES_URL, "rappel_conso", properties=POSTGRES_PROPERTIES  
        )  

        唯一标识列 = "reference_fiche"  

        logging.info("开始处理数据流 ...")  
        query = df_parsed.writeStream.foreachBatch(  
            lambda batch_df, _: (  
                batch_df.join(  
                    existing_data_df, batch_df[唯一标识列] == existing_data_df[唯一标识列], "leftanti"  
                )  
                .write.jdbc(  
                    POSTGRES_URL, "rappel_conso", "append", properties=POSTGRES_PROPERTIES  
                )  
            )  
        ).trigger(once=True) \  
            .start()  

        return query.awaitTermination()

Spark作业的完整代码位于文件src/spark_pgsql/spark_streaming.py中。我们将使用Airflow的DockerOperator来运行此作业，如接下来的章节中所述。

我们来一步步看创建我们所需的Docker镜像以运行Spark作业的过程。这里附上一个Dockerfile供参考，如下所示。

    FROM bitnami/spark:latest  

    WORKDIR /opt/bitnami/spark  

    RUN pip install py4j  

    COPY ./src/spark_pgsql/spark_streaming.py ./spark_streaming.py  
    COPY ./src/constants.py ./src/constants.py  

    ENV POSTGRES_DOCKER_USER=host.docker.internal  
    ARG POSTGRES_PASSWORD  
    ENV POSTGRES_PASSWORD=$POSTGRES_PASSWORD

在该Dockerfile中，我们以bitnami/spark镜像开始，它是一个现成的Spark镜像。接着安装py4j，这是Spark与Python配合使用所需的一个工具。

环境变量 POSTGRES_DOCKER_USER 和 POSTGRES_PASSWORD 用于连接到 PostgreSQL 数据库，POSTGRES_DOCKER_USER 表示用户名，POSTGRES_PASSWORD 则是密码。因为我们的数据库在主机上，我们使用 host.docker.internal 作为主机名。这使我们的 Docker 容器可以访问主机上的服务，例如 PostgreSQL 数据库。PostgreSQL 的密码作为构建参数传递，这样密码就不会硬编码到镜像中。

需要注意的是，这种方法，尤其是在构建时传递数据库密码，可能不足以保证生产环境的安全。这可能会暴露敏感信息。在这种情况下，建议采用更安全的方法，如 Docker BuildKit。

我们现在来为Spark做一个Docker镜像吧。

    docker build -f spark/Dockerfile -t rappel-conso/spark:latest --build-arg POSTGRES_PASSWORD=$POSTGRES_PASSWORD  .

这个命令将构建 rappel-conso/spark:latest 镜像。该镜像包含了运行我们 Spark 作业所需的所有内容，并将用来由 Airflow 的 DockerOperator 执行任务。请在运行该命令时将 $POSTGRES_PASSWORD 替换为您实际的 PostgreSQL 密码。

正如之前所说，Apache Airflow 在数据管道中充当调度工具。它负责调度和管理工作流，确保它们按规定的顺序和条件执行。在我们的系统中，Airflow 用于自动化数据从 Kafka 流处理到 Spark 处理的过程。

我们来看看如下有向无环图（DAG），它将概述任务的序列和依赖关系，让Airflow能够管理它们的执行过程。

    start_date = datetime.today() - timedelta(days=1)  

    default_args = {  
        "owner": "airflow",  
        "start_date": start_date,  
        "retries": 1,  # 失败前的重试次数  
        "retry_delay": timedelta(seconds=5),  
    }  

    with DAG(  
        dag_id="kafka_spark_dag",  
        default_args=default_args,  
        schedule_interval=timedelta(days=1),  
        catchup=False,  
    ) as dag:  

        kafka_stream_task = PythonOperator(  
            task_id="kafka_data_stream",  
            python_callable=stream,  
            dag=dag,  
        )  

        spark_stream_task = DockerOperator(  
            task_id="pyspark_consumer",  
            image="rappel-conso/spark:latest",  
            api_version="auto",  
            auto_remove=True,  
            command="./bin/spark-submit --master local[*] --packages org.postgresql:postgresql:42.5.4,org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0 ./spark_streaming.py",  
            docker_url='tcp://docker-proxy:2375',  
            environment={'SPARK_LOCAL_HOSTNAME': 'localhost'},  
            network_mode="airflow-kafka",  
            dag=dag,  
        )  

        kafka_stream_task 连接至 spark_stream_task

这是配置中的关键元素

• 任务被设定为每天执行。
• 第一个任务是Kafka流任务。它是通过PythonOperator来运行Kafka流函数实现的。此任务将从RappelConso API流式传输数据到Kafka主题，启动数据处理流程。
• 下游任务是Spark流处理任务。它是使用DockerOperator执行的。它运行一个我们定制的Spark Docker容器，负责处理从Kafka接收到的数据。
• 任务是按顺序排列的，其中Kafka流任务先于Spark处理任务。这种顺序很重要，确保数据首先流式传输到Kafka，然后再由Spark处理。

使用 Docker 操作符可以运行对应我们任务的 Docker 容器。这种方法的主要优点包括更方便的包管理和维护、更好的隔离性和更高的可测试性。我们将会通过一个 Spark 流处理任务展示如何使用这个操作符。

以下是spark流处理作业的一些关键docker操作细节。

• 我们将使用在“Spark 设置”部分中提到的镜像 rappel-conso/spark:latest。
• 命令将在容器内运行 Spark 提交命令，指定主节点为本地，并包含 PostgreSQL 和 Kafka 集成所需的必要包，该脚本包含了 Spark 作业的逻辑。
• docker_url 表示运行 Docker 守护进程的主机的 URL。自然的解决办法是将其设置为 unix://var/run/docker.sock，然后在 Airflow Docker 容器中挂载 var/run/docker.sock。我们发现的一个问题是由于权限问题导致无法在 Airflow 容器中使用套接字文件。使用 chmod 777 var/run/docker.sock 更改权限的方法存在严重的安全隐患。为了解决这个问题，我们选择了一个更安全的方案，即采用 bobrik/socat 作为 docker-proxy。这个代理在 Docker Compose 服务中定义，监听 TCP 端口 2375 并将请求转发至 Docker 套接字。

      docker-proxy:  
        image: bobrik/socat  
        command: "TCP4-LISTEN:2375,fork,reuseaddr UNIX-CONNECT:/var/run/docker.sock"  # 命令: 监听TCP 2375端口，并将连接转发到UNIX套接字/var/run/docker.sock  
        ports:  
          - "2376:2375"  # 端口: 将主机上的2376端口映射到容器的2375端口  
        volumes:  
          - /var/run/docker.sock:/var/run/docker.sock  # 卷: 将主机上的/var/run/docker.sock挂载到容器的相同位置  
        networks:  
          - airflow-kafka  # 网络: 加入airflow-kafka网络

在 DockerOperator 里，我们通过 tcp://docker-proxy:2375 这个地址访问主机上的 docker 套接字文件 /var/run/docker.sock，详情可参考 这里 和 这里。

• 最后我们将网络模式设置为 airflow-kafka 模式。这使我们能够使用与代理和运行kafka的Docker相同的网络。这一点至关重要，因为spark作业将从kafka主题中消费数据，因此我们必须确保两个容器能够顺利通信。

在定义了DAG的逻辑之后，我们现在来理解一下docker-compose-airflow.yaml文件中的Airflow服务配置。

Airflow的compose配置文件是基于官方的Apache Airflow docker-compose文件改编的。你可以通过点击这个链接来查看原始文件。

正如该文章所述的这个Airflow版本资源消耗很大，主要是因为核心执行器被设置为CeleryExecutor，这种执行器更适合分布式和大规模的数据处理场景。由于我们的工作负载较小，因此使用单节点的LocalExecutor就已经足够。

这里是我们对airflow的docker-compose的配置所做的改动的简要说明：

• 我们将环境变量 AIRFLOW__CORE__EXECUTOR 设置为 LocalExecutor。
• 我们移除了 airflow-worker 和 flower 服务，因为它们只适用于 Celery 执行器。我们还移除了作为 Celery 后端的 redis 缓存服务。由于我们不会使用 airflow-triggerer，所以我们也把它移除了。
• 我们将 scheduler 和 webserver 的基础镜像 ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.7.3} 替换为一个在运行 docker-compose 时我们将构建的自定义镜像。

    版本: '3.8'  
    x-airflow-common:  # x-airflow-common，用于特定项目配置  
      &airflow-common  # airflow-common，用于特定项目配置  
      构建:  
        上下文: .  
        dockerfile: ./airflow_resources/Dockerfile  # Dockerfile 文件路径  
      镜像: de-project/airflow:latest

• 我们挂载了airflow运行所需的必要卷。AIRFLOW_PROJ_DIR指的是我们将要在后面定义的airflow项目目录。我们还设置了网络为 airflow-kafka ，以便与kafka的引导服务器进行通信。

    # 数据卷挂载点
    volumes:  
      - ${AIRFLOW_PROJ_DIR:-.}/dags:/opt/airflow/dags  
      - ${AIRFLOW_PROJ_DIR:-.}/logs:/opt/airflow/logs  
      - ${AIRFLOW_PROJ_DIR:-.}/config:/opt/airflow/config  
      - ./src:/opt/airflow/dags/src  
      - ./data/last_processed.json:/opt/airflow/data/last_processed.json  
    # 用户ID，默认值为50000，组ID为0
    user: "${AIRFLOW_UID:-50000}:0"  
    # 网络配置
    networks:  
      - airflow-kafka

接下来，我们要创建一些环境变量供 docker-compose 使用。

echo -e "AIRFLOW_UID=$(id -u)\nAIRFLOW_PROJ_DIR=\"./airflow_resources\"" > .env
# 此命令用于设置环境变量 AIRFLOW_UID 和 AIRFLOW_PROJ_DIR

解释：echo -e 命令用于输出字符串到标准输出，这里用于设置环境变量 AIRFLOW_UID 和 AIRFLOW_PROJ_DIR。AIRFLOW_UID 设置为当前用户的ID，AIRFLOW_PROJ_DIR 设置为相对路径 ./airflow_resources。

其中 AIRFLOW_UID 代表 Airflow 容器内的用户ID，AIRFLOW_PROJ_DIR 代表 Airflow项目的目录。

现在一切都准备好了，您可以启动您的Airflow服务了。您可以使用以下命令来启动它：

# Your start command here

     docker compose -f docker-compose-airflow.yaml up

要查看Airflow用户界面，你可以浏览 http://localhost:8080。

Airflow的登录界面。作者供图。

默认情况下，用户名和密码都是 airflow。登录后，您会看到 Airflow 自带的 Dag 列表。查找我们项目中的 kafka_spark_dag 并点击它。

这是Airflow的任务窗口简介，图片由作者提供。

你可以点击任务旁边的按钮开始任务.

接下来，你可以查看任务的状态情况。任务完成时会变成绿色。所以，当所有任务都完成时，它看起来应该像这样：

这张图是由作者画的。

为了确认 rappel_conso_table 是否已有数据，请在 pgAdmin 中运行以下 SQL 语句：

    SELECT count(*) FROM 消费提醒表

当我一月运行这个时，查询返回了10022行。你的结果也应该差不多。

本文成功地展示了使用Kafka、Airflow、Spark、PostgreSQL和Docker构建一个基本但功能完善的数据工程流水线的步骤。主要面向数据工程领域的初学者和新手，它提供了一种动手实践的方法来理解和实现数据流处理和存储的关键概念。

在这份指南中，我们详细介绍了流水线的每个组件，从设置Kafka进行数据流到使用Airflow进行任务调度，再到使用Spark处理数据，并将其存储在PostgreSQL数据库中。在项目中使用Docker简化了设置过程，并确保了在不同环境中的稳定性。

需要注意的是，虽然这种设置对于学习和小型项目来说非常理想，但要将其扩展到生产使用，则需要考虑更多的因素，特别是在安全性和性能优化方面。未来的改进可以包括集成更高级的数据处理技术，探索实时分析功能，甚至扩展管道以纳入更多复杂的数据源。

说到底，这个项目为那些希望亲自上手数据处理的人提供了一个实用的起点。它帮助理解基础知识，为进一步探索该领域提供了稳固的基础。

在第二部分，我们将探讨如何有效地利用存储在我们PostgreSQL数据库中的数据。我们将介绍由大型语言模型（LLMs）驱动的代理程序以及一系列工具，帮助我们用自然语言查询与数据库交互。敬请期待！

• 领英 : https://www.linkedin.com/in/hamza-gharbi-043045151/
• 推特 : https://twitter.com/HamzaGh25079790