ByteHouse+Apache Airflow:高效简化数据管理流程_今日讯
Apache Airflow 与 ByteHouse 相结合,为管理和执行数据流程提供了强大而高效的解决方案。本文突出了使用 Apache Airflow 与 ByteHouse 的主要优势和特点,展示如何简化数据工作流程并推动业务成功。
(资料图)
主要优势
可扩展可靠的数据流程:Apache Airflow 提供了一个强大的平台,用于设计和编排数据流程,让您轻松处理复杂的工作流程。搭配 ByteHouse,一款云原生的数据仓库解决方案,您可以高效地存储和处理大量数据,确保可扩展性和可靠性。
自动化工作流管理:Airflow 的直观界面通过可视化的 DAG(有向无环图)编辑器,使得创建和调度数据工作流程变得容易。通过与 ByteHouse 集成,您可以自动化提取、转换和加载(ETL)过程,减少手动工作量,实现更高效的数据管理。
简单的部署和管理:Apache Airflow 和 ByteHouse 均设计为简单的部署和管理。Airflow 可以部署在本地或云端,而 ByteHouse 提供完全托管的云原生数据仓库解决方案。这种组合使得数据基础设施的设置和维护变得无缝化。
客户场景
业务场景
在这个客户场景中,一家名为“数据洞察有限公司(假名)”的分析公司,他们将 Apache Airflow 作为数据管道编排工具。他们选择 ByteHouse 作为数据仓库解决方案,以利用其强大的分析和机器学习功能。
数据洞察有限公司在电子商务行业运营,并收集存储在 AWS S3 中的大量客户和交易数据。他们需要定期将这些数据加载到 ByteHouse,并执行各种分析任务,以获得对业务运营的洞察。
数据链路
使用 Apache Airflow,数据洞察有限公司设置了一个基于特定事件或时间表的数据加载管道。例如,他们可以配置 Airflow 在每天的特定时间触发数据加载过程,或者当新的数据文件添加到指定的 AWS S3 存储桶时触发。当触发事件发生时,Airflow 通过从 AWS S3 中检索相关数据文件来启动数据加载过程。它使用适当的凭据和 API 集成确保与 S3 存储桶的安全身份验证和连接。一旦数据从 AWS S3 中获取,Airflow 会协调数据的转换和加载到 ByteHouse 中。它利用 ByteHouse 的集成能力,根据预定义的模式和数据模型高效地存储和组织数据。
成功将数据加载到 ByteHouse 后,数据洞察有限公司可以利用 ByteHouse 的功能进行分析和机器学习任务。他们可以使用 ByteHouse 的类 SQL 语言查询数据,进行复杂的分析,生成报告,并揭示有关客户、销售趋势和产品性能的有意义洞察。
此外,数据洞察有限公司还利用 ByteHouse 的功能创建交互式仪表板和可视化。他们可以构建动态仪表板,显示实时指标,监控关键绩效指标,并与组织中的利益相关者共享可操作的洞察。
最后,数据洞察有限公司利用 ByteHouse 的机器学习功能来开发预测模型、推荐系统或客户细分算法。ByteHouse 提供了必要的计算能力和存储基础设施,用于训练和部署机器学习模型,使数据洞察有限公司能够获得有价值的预测性和规定性洞察。
总结
通过使用 Apache Airflow 作为数据管道编排工具,并将其与 ByteHouse 集成,数据洞察有限公司实现了从 AWS S3 加载数据到 ByteHouse 的流畅自动化流程。他们充分利用 ByteHouse 的强大分析、机器学习和仪表板功能,获得有价值的洞察,并推动组织内的数据驱动。
ByteHouse<>AirFlow 快速入门
先决条件
在您的虚拟/本地环境中安装 pip。在您的虚拟/本地环境中安装 ByteHouse CLI 并登录到 ByteHouse 账户。参考 ByteHouse CLI 以获取安装帮助。macOS 上使用 Homebrew 的示例brew install bytehouse-cli
安装 Apache Airflow
在本教程中,我们使用 pip 在您的本地或虚拟环境中安装 Apache Airflow。了解更多信息,请参阅官方 Airflow 文档。
# airflow需要一个目录,~/airflow是默认目录,# 但如果您喜欢,可以选择其他位置#(可选)export AIRFLOW_HOME=~/airflowAIRFLOW_VERSION=2.1.3PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)"# 例如:3.6CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"
如果使用 pip 无法安装,请尝试使用 pip3 install 进行安装。安装完成后,运行命令 airflow info 以获取有关 Airflow 的更多信息。
Airflow 初始化
通过执行以下命令来初始化 Airflow 的 Web 服务器
# 初始化数据库airflow db initairflow users create \--username admin \--firstname admin \--lastname admin \--role Admin \--email admin# 启动Web服务器,默认端口是8080# 或修改airflow.cfg设置web_server_portairflow webserver --port 8080
设置好 Web 服务器后,您可以访问 http://localhost:8080/使用先前设置的用户名和密码登录 Airflow 控制台。
在新的终端中,使用以下命令设置 Airflow 调度器。然后,刷新 http://localhost:8080/。
YAML 配置
使用 cd ~/airflow 命令进入 Airflow 文件夹。打开名为 airflow.cfg 的配置文件。添加配置并连接到数据库。默认情况下,您可以使用 SQLite,但也可以连接到 MySQL。
# 默认情况下是SQLite,也可以连接到MySQLsql_alchemy_conn = mysql+pymysql://airflow:airflow@xxx.xx.xx.xx:8080/airflow# authenticate = False# 禁用Alchemy连接池以防止设置Airflow调度器时出现故障 https://github.com/apache/airflow/issues/10055sql_alchemy_pool_enabled = False# 存放Airflow流水线的文件夹,通常是代码库中的子文件夹。该路径必须是绝对路径。dags_folder = /home/admin/airflow/dags
创建有向无环图(DAG)作业
在 Airflow 路径下创建一个名为 dags 的文件夹,然后创建 test_bytehouse.py 以启动一个新的 DAG 作业。
~/airflowmkdir dagscd dagsnano test_bytehouse.py
在 test_bytehouse.py 中添加以下代码。该作业可以连接到 ByteHouse CLI,并使用 BashOperator 运行任务、查询或将数据加载到 ByteHouse 中。
from datetime import timedeltafrom textwrap import dedentfrom airflow import DAGfrom airflow.operators.bash import BashOperatorfrom airflow.utils.dates import days_agodefault_args = { "owner": "airflow", "depends_on_past": False, "email": ["airflow@example.com"], "email_on_failure": False, "email_on_retry": False, "retries": 1, "retry_delay": timedelta(minutes=5),}with DAG( "test_bytehouse", default_args=default_args, description="A simple tutorial DAG", schedule_interval=timedelta(days=1), start_date=days_ago(1), tags=["example"],) as dag: tImport = BashOperator( task_id="ch_import", depends_on_past=False, bash_command="$Bytehouse_HOME/bytehouse-cli -cf /root/bytehouse-cli/conf.toml "INSERT INTO korver.cell_towers_1 FORMAT csv INFILE \"/opt/bytehousecli/data.csv\" "", ) tSelect = BashOperator( task_id="ch_select", depends_on_past=False, bash_command="$Bytehouse_HOME/bytehouse-cli -cf /root/bytehouse-cli/conf.toml -q "select * from korver.cell_towers_1 limit 10 into outfile \"/opt/bytehousecli/dataout.csv\" format csv "" ) tSelect >> tImport
在当前文件路径下运行 python test_bytehouse.py 以在 Airflow 中创建 DAG。在浏览器中刷新网页。您可以在 DAG 列表中看到新创建的名为 test_bytehouse 的 DAG。
执行 DAG
在终端中运行以下 Airflow 命令来查看 DAG 列表和 test_bytehouse DAG 中的子任务。您可以分别测试查询执行和数据导入任务。
#打印"test_bytehouse" DAG中的任务列表[root@VM-64-47-centos dags]# airflow tasks list test_bytehousech_importch_select#打印"test_bytehouse" DAG中任务的层次结构[root@VM-64-47-centos dags]# airflow tasks list test_bytehouse --tree
运行完 DAG 后,查看您的 ByteHouse 账户中的查询历史页面和数据库模块。您应该能够看到查询/加载数据成功执行的结果。
关键词:
您可能也感兴趣:
为您推荐
松原市举行“筑梦新时代 悦动向未来”教育系统专场文艺演出 天天快播报
人过五十,低配生活,高配格局 天天微资讯
全球观天下!《星之海洋2》重制版等年末Switch游戏容量确定
排行
最近更新
- ByteHouse+Apache Airflow:高效简化数据管理流程_今日讯
- 国际禁毒日丨毒贩在哪里,我们的战场就在哪里
- 环球速读:注意防范!南宁大部分中小河流出现0.5~1.7米涨幅
- 新华全媒+|百万千瓦时绿电将服务2023年夏季达沃斯论坛
- 几年前,我存在朋友公司的20万本金,今天终于拿回来了。
- 世界快播:河南宝丰:“水电气暖网有线电视6合1”一次搞定
- 俄罗斯莫斯科市、莫斯科州和沃罗涅日州反恐制度状态已解除-当...
- 科技“关键变量”撬动产业“最大增量”——三亚崖州湾科技城...
- 当前快报:济南5G基站4.3万处、上云企业超6万家,均居全省首位
- 战火与永恒怎么提高统领值 战火与永恒统御力提升攻略
- 焦点速递!全国多个口岸客流高位运行 多措施保障出入境旅客...
- 农业农村部:6月26日"农产品批发价格200指数"比昨天下降0.36个点
- 车谷税务:“定制服务”助力这家企业跻身全国前五 环球视讯
- 三六零(601360)6月26日14点41分触及跌停板 今头条
- 2023浙江科技学院志愿填报招生计划数据参考(省内招生+省外招生)
- 星火成炬 | 点亮乡村多彩生活 全球讯息
- 全球微头条丨2023宁波宁海中考普高投档线划定
- 储能行业发展的六大趋势 环球观点
- 江苏高邮农商银行携手高邮供销合作总社助力乡村振兴
- 视点!助力形成船员维权共商机制 厦门船员权益保障工作室正...
- 2023年保定中考志愿填报时间+规则
- 冬天豆豉怎么捂好发酵? 今日视点
- 环球讯息:2023年河北高考艺术类录取分数线
- 天天报道:陕西省宝塔区发布雷雨大风黄色预警
- 2023郑州小升初摇号学校名单 世界新消息
- 热资讯!郑州人才公寓6月第三批都有哪些房源可以申请?
- 全球要闻:蔡徐坤演唱会互动女网红道歉 基本情况讲解
- 天天简讯:全网超4000万预约!《逆水寒》手游现已开启下载
- 合肥市凤凰城小学报名咨询电话
- 嫩江市 坚定不移增进人民福祉