标签：airflow hour 自定义 partition HivePartitionSensor hive apache table

HivePartitionSensor功能

用于检查hive表的分区是否存在，在某些场景下可以使用该sensor来替代ExternalTaskSensor，且使用起来更加便捷。

场景描述

在数仓中这么两种表：
表1 ：daily_table，该表是按天分区的表，一天跑一次。
表2 ：hour_table，该表是按照小时分区，每个小时跑一次。
其中daily_table 依赖于 hour_table的执行，对于这种执行周期不同的任务，如果通过airflow的ExternalTaskSensor来声明它们之间的依赖会非常麻烦，为此可以使用HivePartitionSensor来解决

代码如下

简版：

	# 用于检查小时级任务，每天23点的分区
	check_hour_table= HivePartitionSensor(
        task_id='check_task',
        metastore_conn_id='hive-conn',		# hive的hive_metastore连接，可点击ariflow web界面的Connection进行配置
        table='库名.hour_table',		# 需要检查的hive表名。注意：需要加上数据库名。
        mode='reschedule',					# reschedule: 该模式在休眠期间不会占用slot，只有在执行时才会占用
        poke_interval=300,					# 两次检查的间隔时间，单位秒。使用reschedule模式时，建议该值不小于60。
        partition='year=2021 and month=12 and day=31 and hour=23',  # 需要检测的分区,分区格式需要实际情况
        # timeout=600,						# 超时时间，单位秒。可根据情况选择是否使用。
		# soft_fail=false,					# 如果设置为true，则失败时将任务标记为跳过。默认false
    )
    
    # 天级任务
    daily_table = DummyOperator(
        task_id='server_db_member_wallet_consumable_tx_merge',
    )
    check_hour_table >> daily_table

详版

from airflow import DAG
from datetime import datetime, timedelta
from airflow.providers.apache.hive.sensors.hive_partition import HivePartitionSensor
# 该方法用于将时间格式化为hive表的分区格式。其中interval=9，是用来转换时区的。logical_date默认是utc时区
def get_hour_partition(logical_date, hour, interval=9):
    new_date = (logical_date + timedelta(hours=interval))
    y = str(new_date)[0:4]
    m = str(new_date)[5:7]
    d = str(new_date)[8:10]
    h = str(hour)
    partition = 'year=%s and month=%s and day=%s and hour=%s' % (y, m, d, h)
    return partition


# [START instantiate_dag]
with DAG(
        dag_id='spark_sql',
        schedule_interval=None,
        user_defined_macros={
            "get_hour_partition": get_hour_partition,
        }, # 定义get_hour_partition方法无法直接被airflow的sensor或operator使用，需要将其注册为'宏变量'，(自定义宏变量)
) as dag:
	# 用于检查小时级任务，每天23点的分区
	check_hour_table= HivePartitionSensor(
        task_id='check_task',
        metastore_conn_id='hive-conn',
        table='库名.hour_table',
        mode='reschedule',
        poke_interval=300,
        partition='{{ get_hour_partition(logical_date,23) }}',  # 调用自定义的方法格式化分区
    )
    
    # 天级任务
    daily_table = DummyOperator(
        task_id='server_db_member_wallet_consumable_tx_merge',
    )
    check_hour_table >> daily_table

注意

1. 需要装的python module：apache-airflow-providers-apache-hive
2. 导包：from airflow.providers.apache.hive.sensors.hive_partition import HivePartitionSensor

参考文档

HivePartitionSensor
https://airflow.apache.org/docs/apache-airflow-providers-apache-hive/stable/_api/airflow/providers/apache/hive/sensors/hive_partition/index.html
BaseSensorOperator
https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/sensors/base/index.html#module-airflow.sensors.base

说明： BaseSensorOperator中的参数在所有Sensor中都通用，包括HivePartitionSensor

标签：airflow,hour,自定义,partition,HivePartitionSensor,hive,apache,table
来源： https://blog.csdn.net/qq1277419839/article/details/122261051

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。