---
title: "Airflow 集成 OceanBase 数据库 - OceanBase 数据库 V4.3.5 | OceanBase 文档中心"
description: Airflow 集成 OceanBase 数据库 Apache Airflow 是一个开源平台，用于开发、调度和监控面向批处理的工作流程。Airflow 所有的工作流程都可以使用 Python 代码定义。Web 界面可以管理工作流程的状态。 前提条件 已安装 Apache Airflow，详细说明，请参考 Apach…
---
切换语言

- 中文站 - 简体中文
- International - English
- 日本站 - 日本語

文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*P8CuR4UJ_FkAAAAAAAAAAAAADiGDAQ/original) OceanBase 数据库分布式版 - V 4.3.5 LTS

# Airflow 集成 OceanBase 数据库

更新时间：2026-04-09 14:12:04

[编辑](https://github.com/oceanbase/oceanbase-doc/edit/V4.3.5/zh-CN/680.ecological-integration/600.orchestration/100.airflow.md)  

Apache Airflow 是一个开源平台，用于开发、调度和监控面向批处理的工作流程。Airflow 所有的工作流程都可以使用 Python 代码定义。Web 界面可以管理工作流程的状态。

## 前提条件

- 已安装 Apache Airflow，详细说明，请参考 [Apache Airflow 官网](https://airflow.apache.org/)。
 - 已安装 OceanBase 数据库并且创建了 MySQL 模式租户。
 - 确保待连接的 OBServer 节点 IP 与 Airflow 所在机器保持网络连通。

## 获取数据库连接信息

联系 OceanBase 数据库部署人员或者管理员获取相应的数据库连接串，例如：

```shell
obclient -h$host -P$port -u$user_name -p$password -D$database_name

```

**参数说明：**

- `$host`：提供 OceanBase 数据库连接 IP。OceanBase 数据库代理（OceanBase Database Proxy，ODP）连接方式使用的是一个 ODP 地址；直连方式使用的是 OBServer 节点的 IP 地址。
 - `$port`：提供 OceanBase 数据库连接端口。ODP 连接的方式默认是 `2883`，在部署 ODP 时可自定义；直连方式默认是 `2881`，在部署 OceanBase 数据库时可自定义。
 - `$database_name`：需要访问的数据库名称。
 - `$user_name`：提供租户的连接账户。ODP 连接的常用格式：`用户名@租户名#集群名` 或者 `集群名:租户名:用户名`；直连方式格式：`用户名@租户名`。
 - `$password`：提供账户密码。

#### 注意

连接租户的用户需要拥有该数据库的 `CREATE`、`INSERT`、`DROP` 和 `SELECT` 权限。

**示例如下：**

```shell
obclient -hxxx.xxx.xxx.xxx -P2881 -utest_user001@mysql001 -p****** -Dtest

```

## 在 Airflow 中添加 OceanBase 数据源进行连接

1. 打开 Airflow Web UI。
 2. 导航到 Admin -> Connections。
 3. 点击 "+" 号来添加一个新的连接。
 4. 填写以下字段：

   | 配置项 | 说明 |
   | --- | --- |
   | Connection Id | ob（可以是任何标识符）。 |
   | Connection Type | MySQL |
   | Host | 取自连接串中 `-h` 参数，OceanBase 数据库连接 IP 地址，例如 `xxx.xxx.xxx.xxx`。 |
   | Schema | 取自连接串中 `-D` 参数，需要访问的数据库名称。 |
   | Login | 取自连接串中 `-u` 参数，账号名称，例如 `test_user001@mysql001`。 |
   | Password | 取自连接串中 `-p` 参数，账号密码。 |
   | Port | 取自连接串中 `-P` 参数，OceanBase 数据库连接端口，直连方式默认为 `2881`，通过 ODP 连接默认为 `2883`。 |
 5. 创建成功后，即可在 Airflow 任务中通过引用 `ob` 这个 Connection Id，来访问 OceanBase 数据库。

## Airflow 任务示例

在 Airflow 中添加 OceanBase 数据库之后，可以通过编写以下内容，实现 AirFlow 从 OceanBase 数据库中读取数据并打印。

1. 在 Airflow 安装目录下的 dags 文件夹中新建 query.py 文件，并编辑如下内容。

   ```Python
   from airflow import DAG
   from airflow.utils.dates import days_ago
   from airflow.providers.mysql.hooks.mysql import MySqlHook
   from airflow.operators.python import PythonOperator

   default_args = {
       'owner': 'airflow',
       'retries': 0,
   }

   def fetch_and_print_data():
       hook = MySqlHook(mysql_conn_id='ob')
       sql = "SELECT * FROM person LIMIT 1;"
       connection = hook.get_conn()
       cursor = connection.cursor()
       cursor.execute(sql)
       rows = cursor.fetchall()
       for row in rows:
           print(row)

   with DAG(
       dag_id='sql_query',
       default_args=default_args,
       schedule_interval='@daily',
       start_date=days_ago(1),
       catchup=False,
   ) as dag:

       run_and_print = PythonOperator(
           task_id='run_and_print',
           python_callable=fetch_and_print_data,
       )

       run_and_print

   ```
 2. 执行 `airflow tasks test sql_query run_and_print`，即可打印 person 表中的第一条数据。

 上一篇 下一篇 ![有帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*y6ocSqN8cqsAAAAAAAAAAAAAARQnAQ)![无帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*BG9IQJyLHF8AAAAAAAAAAAAAARQnAQ)![反馈](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*eTWdQKCRKHwAAAAAAAAAAAAAARQnAQ)[AI](https://www.oceanbase.com/obi) 咨询热线