OpenMLDB Airflow Connector：让 MLOps 工作流更自由

导读

为提升 OpenMLDB 的易用性、便捷性，降低开发者的使用门槛，打造更顺滑高效的MLOps工作流，OpenMLDB 一直以来都在积极增进社区间合作，自觉生态上下游打通。今天的这篇文章要和大家介绍的就是 OpenMLDB 又一新生态组件——OpenMLDB Airflow Connector！这个工具是 OpenMLDB 和流行的生产级调度编排系统 Apache Airflow 积极整合的产物，也将是方便开发者打造端到端 MLOps 工作流的另一利器。

接下来 OpenMLDB PMC 黄威会为大家简要介绍 OpenMLDB Airflow Connector 并演示其部署使用流程。

OpenMLDB Airflow Connector 优势

OpenMLDB Airflow Connector 链接了 OpenMLDB 和 Airflow，使开发者可以在 Airflow 中直接使用 OpenMLDB，不仅在操作上便捷易用，更重要的是集合了两大产品的优势，碰撞出了更亮眼的火花：

• 使用简单易于管理：Airflow 简洁大方的 UI 界面和丰富的命令工具帮你告别部署测试的繁琐，而 OpenMLDB 以 SQL 为核心的开发管理体系助力开发者进一步降低学习成本和使用门槛，更加易用易管。
• AI流程自动化处理：OpenMLDB＋Airflow打造端到端的MLOps工作流，让数据和特征的处理和流动更加自动化，帮助开发者远离调试异常、专注核心业务，更加接近开发即上线的目标。
• 灵活性强自由度高：OpenMLDB 支持用户自定义函数，而 Airflow 也欢迎开发者用代码构建任何需要的感应器和处理工具，将 OpenMLDB 快速接入 Airflow 后整个机器学习流程的自由度被进一步开发。开发者应用更加灵活自由，可操作性极大增强。

流程展示

关键步骤

我们使用 Airflow dag 完成 talking data demo 中的特征计算与上线，并使用 talking data demo 的 predict server 来进行上线后的实时推理测试。

1. 下载 demo 用文件
2. 启动 OpenMLDB，predict server，airflow standalone
3. airflow web 页面中创建 connection，运行 dag
4. predict 测试

具体展示

下载 demo 用的文件

wget https://openmldb.ai/download/airflow_demo/airflow_demo_files.tar.gztar zxf airflow_demo_files.tar.gzls airflow_demo_files
1

运行容器，并暴露 Airflow web 端口，以及将文件映射到 docker 容器内。

docker run -p 8080:8080 -v `pwd`/airflow_demo_files:/work/airflow/dags -it 4pdosc/openmldb:0.6.0 bash
1

OpenMLDB 镜像中还未包括 Airflow 与 provider，所以需要下载 Airflow 与 OpenMLDB provider，再启动 Airflow。

pip3 install airflow-provider-openmldb
1

源数据准备，example dag 中的源数据放在 /tmp 目录，我们直接将数据拷贝到 /tmp 使用，避免修改 dag 文件。

cp /work/talkingdata/train_sample.csv /tmp/
1

启动 OpenMLDB。

/work/init.sh
1

启动 predict server，使用 /work/talkingdata/ 中的 predict server。

python3 /work/talkingdata/predict_server.py --no-init > predict.log 2>&1 &
1

启动 Airflow，演示可以使用 standalone 模式。

export AIRFLOW_HOME=/work/airflowcd /work/airflowairflow standalone 
1

Airflow standalone 运行输出将提示登录用户名和密码，例如

登录 Airflow Web 界面 http://localhost:8080 ，并输入用户名和密码。

dag 将被自动加载，点击 example_openmldb_complex，可以点击 Code 查看 dag 的详情。

可以看到 OpenMLDB operator 所需 connection 的 id 为 openmldb_conn_id。

创建 connection openmldb_conn_id。在管理界面中点击 connections，并添加 connection。

Airflow OpenMLDB Provider 是连接 OpenMLDB Api Server 的，所以此处配置中填入OpenMLDB Api Server 的地址，而不是 zookeeper 地址。具体配置如下所示：

创建完成后的 connection 如下图所示：

运行dag，即完成一次训练模型、sql部署与模型部署。成功运行的结果，类似下图：

测试：

导入在线数据

Airflow DAG 中完成了 SQL 和模型的上线。但在线存储中还没有数据，所以我们需要做一次在线数据导入。

curl -X POST http://127.0.0.1:9080/dbs/example_db -d'{"mode":"online", "sql":"load data infile \"file:///tmp/train_sample.csv\" into table example_table options(mode=\"append\");"}'
1

这是一个异步操作，但由于数据量小，也会很快完成。通过SHOW JOBS也可以查看导入操作的状态。

curl -X POST http://127.0.0.1:9080/dbs/example_db -d'{"mode":"online", "sql":"show jobs"}'
1

执行 python3 /work/talkingdata/predict.py 进行一次预测，预测将使用新部署好的 SQL 与模型。

安装部署

pip install airflow-provider-openmldb

总结

本文为大家介绍了 OpenMLDB Airflow Connector，并展示了它的部署安装和操作运行流程，希望能够帮助大家了解 OpenMLDB Airflow Connector 功能作用和优势特点。

期待各位持续关注 OpenMLDB，OpenMLDB 社区将持续打造面向上下游技术组件的生态圈，为开发者降低使用门槛，提供更多简单易用的连接工具。

• 面向线上数据生态，如 Pulsar（已完成），Kafka（已完成），RocketMQ（已完成），Flink，RabbitMQ 等

• 面向离线数据生态，如 HDFS，HBase，Cassandra，S3 等

• 面向模型构建的算法、框架，如 XGBoost，LightGBM，TensorFlow，PyTorch，Scikit Learn 等

• 面向机器学习建模全流程的调度框架、部署工具，如 DolphinScheduler（已完成），Airflow（已完成），Byzer（已完成），Kubeflow，Prometheus，Grafana 等

相关链接

OpenMLDB github 主页： https://github.com/4paradigm/OpenMLDB

Airflow 官网： https://airflow.apache.org/

Airflow github 主页： https://github.com/apache/airflow

OpenMLDB 微信交流群