成都公司核名的网站,网站开发和数据库哪个有前途,苏州网站维护,宿州网站推广#x1f3e1; 个人主页#xff1a;IT贫道_大数据OLAP体系技术栈,Apache Doris,Clickhouse 技术-CSDN博客 #x1f6a9; 私聊博主#xff1a;加入大数据技术讨论群聊#xff0c;获取更多大数据资料。 #x1f514; 博主个人B栈地址#xff1a;豹哥教你大数据的个人空间-豹… 个人主页IT贫道_大数据OLAP体系技术栈,Apache Doris,Clickhouse 技术-CSDN博客 私聊博主加入大数据技术讨论群聊获取更多大数据资料。 博主个人B栈地址豹哥教你大数据的个人空间-豹哥教你大数据个人主页-哔哩哔哩视频 目录
1. HiveOperator配置
2. HiveOperator调度HQL案例 1. HiveOperator配置
可以通过HiveOperator直接操作Hive SQL HiveOperator的参数如下
hql(str):需要执行的Hive SQL。hive_cli_conn_id(str):连接Hive的conn_id在airflow webui connection中配置的。
想要在airflow中使用HiveOperator调用Hive任务首先需要安装以下依赖并配置Hive Metastore
#切换Python37环境[rootnode4 ~]# conda activate python37#安装hive provider package(python37) [rootnode4 ~]# pip install apache-airflow-providers-apache-hive2.0.2#启动airflow(python37) [rootnode4 ~]# airflow webserver --port 8080(python37) [rootnode4 ~]# airflow scheduler
登录Airflow webui并设置Hive Metastore登录后找到”Admin”-”Connections”点击“”新增配置 2. HiveOperator调度HQL案例
1 启动Hive准备表
启动HDFS、Hive Metastore在Hive中创建以下三张表
create table person_info(id int,name string,age int) row format delimited fields terminated by \t;create table score_info(id int,name string,score int) row format delimited fields terminated by \t;
向表 person_info加载如下数据
1 zs 182 ls 193 ww 20
向表score_info加载如下数据
1 zs 1002 ls 2003 ww 300
2) 在node4节点配置Hive 客户端
由于Airflow 使用HiveOperator时需要在Airflow安装节点上有Hive客户端所以需要在node4节点上配置Hive客户端。
将Hive安装包上传至node4 “/software”下解压并配置Hive环境变量
#在/etc/profile文件最后配置Hive环境变量export HIVE_HOME/software/hive-1.2.1export PATH$PATH:$HIVE_HOME/bin#使环境变量生效source /etc/profile
修改HIVE_HOME/conf/hive-site.xml 写入如下内容
configurationpropertynamehive.metastore.warehouse.dir/namevalue/user/hive/warehouse/value/propertypropertynamehive.metastore.local/namevaluefalse/value/propertypropertynamehive.metastore.uris/namevaluethrift://node1:9083/value/property/configuration
3 编写DAG python配置文件
注意在本地开发工具编写python配置时需要用到HiveOperator需要在本地对应的python环境中安装对应的provider package。
C:\Users\wubaid:D:\cd d:\ProgramData\Anaconda3\envs\python37\Scriptsd:\ProgramData\Anaconda3\envs\python37\Scriptspip install apache-airflow-providers-apache-hive2.0.2注意这里本地安装也有可能缺少对应的C环境我们也可以不安装直接跳过也可以。
Python配置文件:
from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.apache.hive.operators.hive import HiveOperatordefault_args {owner:wangwu,start_date:datetime(2021, 9, 23),retries: 1, # 失败重试次数retry_delay: timedelta(minutes5) # 失败重试间隔
}dag DAG(dag_id execute_hive_sql,default_argsdefault_args,schedule_intervaltimedelta(minutes1)
)firstHiveOperator(task_idperson_info,hive_cli_conn_idnode1-hive-metastore,hqlselect id,name,age from person_info,dag dag
)secondHiveOperator(task_idscore_info,hive_cli_conn_idnode1-hive-metastore,hqlselect id,name,score from score_info,dagdag
)thirdHiveOperator(task_idjoin_info,hive_cli_conn_idnode1-hive-metastore,hqlselect a.id,a.name,a.age,b.score from person_info a join score_info b on a.id b.id,dagdag
)first second third
4) 调度python配置脚本
将以上配置好的python文件上传至node4节点$AIRFLOW_HOME/dags下重启Airflow websever与scheduler登录webui开启调度 调度结果如下
