Airflow HiveServer2Hook LDAP 연결

HiveServer2 인증 방법으로 LDAP을 사용하는 경우

Airflow HiveServer2Hook를 사용해 hive에 query을 날리는 방법에 대해서 알아보도록 하자

Airflow의 버전은 2.0.0을 사용했다.

 

먼저 Airflow의 HiveServer2Hook의 코드를 살펴보도록 하자

github.com/apache/airflow/blob/c699e97d0d43e06ea9835a845570db4fc7fd6931/airflow/providers/apache/hive/hooks/hive.py#L804

 

코드에서 살펴봐야 할 부분은 인증방식에 대한 부분과 connection을 만드는 과정이다.

먼저 인증 방식을 설정하기 위해서는 829번 line의 코드를 살펴봐야 한다

auth_mechanism으로 connection에 있는 authMechanism의 데이터를 이용한다

우리는 LDAP을 이용할 것이기 때문에 LDAP 부분을 확인하고 인증을 위해 password까지 필요한 것을 확인하면 된다

 

다음으로 확인할 부분은 return으로 pyhive의 connect인데 이 부분에서 오류가 발생할 수 있다

pyhive에서 SASL 관련 에러가 발생하는 경우 이전 글을 확인해보자

2021.04.22 - [개발/python] - pyHive Hive LDAP SASL error

이제 HiveServer2 연결을 위한 Connection을 만들어보자

Conn Type는 Hive Server 2 Thrift로 설정하고,

코드에서 확인한대로 authMechanism을 LDAP로 설정하기 위해 Extra에 해당 내용을 추가한다

{"authMechanism":"LDAP"}

이제 HiveServer2Hook을 사용하는 dag를 만들어보자 

github.com/dydwnsekd/airflow_example/blob/main/dags/HiveServer2Hook_auth_LDAP.py

from airflow import DAG
from datetime import datetime, timedelta
from airflow.providers.apache.hive.hooks.hive import *
from airflow.operators.python_operator import PythonOperator

default_args = {
    'start_date': days_ago(1),
    'retries': 0,
    'catchup': False,
    'retry_delay': timedelta(minutes=5),
}

def simple_query():
    hql = "SELECT * FROM airflow LIMIT 10"
    hm = HiveServer2Hook(hiveserver2_conn_id = 'HiveServer2_test')
    result = hm.get_records(hql, "db_name")
    
    print(result)
    
    for i in result:
    	print(i)
        
dag = DAG(
        'hivehook_test',
    	default_args=default_args,
    	scheduler_interval="@once",
    )
    
t1 = PythonOperator(
    task_id = 'HiveServer2Hook_test',
    python_callable=simple_query,
    dag=dag,
)

이제 실행을 시켜보자

결과로는 아래와 같이 data1, data2 로 동일한 데이터가 2 row 나왔다

실제 Hive에 저장된 데이터도 아래와 같다

 

이렇게 Airflow HiveServer2Hook를 사용해 Hive LDAP인증방식에 접근하는 방법을 알아보았다.

다음 글에서는 HiveOperator을 사용해 Hive에 접근해보도록 하자