如何用Docker部署的Apache Airflow自动化本地Apache Kafka的Python生产消费任务？

一、Docker Airflow 连接本地机器 Kafka #

1. 调整 Kafka 网络配置 #

Kafka 默认仅绑定 localhost，Docker 容器无法直接访问，需修改 server.properties 配置文件：

• 设置 listeners=PLAINTEXT://0.0.0.0:9092，允许所有网卡监听 9092 端口
• 设置 advertised.listeners=PLAINTEXT://<你的本地IP>:9092，替换成本地真实IP（如 192.168.1.100，不能用 localhost）
• 重启 Kafka 服务使配置生效

2. 让 Airflow 容器访问本地 Kafka #

有两种可行方式：

• Host 网络模式启动
  启动 Airflow 时加上 --network host 参数，容器直接复用宿主机网络，Python 脚本中直接用 <本地IP>:9092 连接 Kafka。示例命令：

  docker-compose up -d --network host
  

  注意：此模式下容器端口会直接占用宿主机端口，需提前确认无冲突。

• 指定本地IP连接
  不使用 host 网络的话，在 Airflow 的 Python 任务代码中，将 Kafka bootstrap servers 设置为 <本地IP>:9092，同时确保本地防火墙开放 9092 端口，允许 Docker 所在网络访问。

3. 验证连接有效性 #

在 Airflow PythonOperator 中编写测试代码，确认连接正常：

from kafka import KafkaProducer

def send_test_message():
    producer = KafkaProducer(bootstrap_servers=['192.168.1.100:9092'])
    producer.send('test_topic', b'Message from Airflow Docker')
    producer.flush()
    print("Test message sent to Kafka")

运行任务后，用本地 Kafka 消费者查看是否收到消息。

二、AWS EC2 部署 Airflow 端口冲突解决 #

1. 排查端口占用 #

登录 EC2 实例，执行命令查看 Airflow 默认依赖端口（8080、5432、6379 等）的占用情况：

sudo netstat -tulpn | grep -E '8080|5432|6379'

或使用 ss 命令：

sudo ss -tulpn | grep -E '8080|5432|6379'

定位占用端口的进程，判断是否为其他服务占用。

2. 修改 Airflow 端口配置 #

若确认是端口冲突，修改 airflow.cfg 配置文件：

• 将 webserver 端口改为未被占用的端口：web_server_port = 8081
• 若使用 PostgreSQL、Redis 等依赖，也需对应调整这些服务的端口，或停止占用端口的无关服务。

3. 配置安全组与防火墙 #

EC2 实例的安全组需开放新端口的入站规则，同时实例内部防火墙（如 ufw、firewalld）也要允许该端口的访问请求。

内容的提问来源于stack exchange，提问作者Abdul Moiz