[Airflow] HiveServer2Hook, HiveCliHook 사용하여 Hive에 접근하기

 

2023.06.12 (MON) 학습정리
#Airflow #Hive #HiveServer2Hook #HiveCliHook

---

지난 글에서 단순히 HiveOperator를 활용하여 단순히 query를 전달하는 방식으로 hive에 접근하였으나,

이번에는 HiveServer2를 통해  Hiveserver2Hook,  HiveCliHook을 활용하고자함 

 

1. HiverServer2

HiveServer2는 클라이언트가 하이브에 대해 쿼리를 실행할 수 있게 해주는 서비스로 HiveServer2Hook, HiveCliHook 사용 전 HiverServer2에 접근하기 위한 기본 설정을 진행해야 함

• os 계정명 (username) 확인

$ hdfs dfs -ls -R /user/hive
drwxr-xr-x   - username supergroup          0 2023-06-12 12:31 /user/hive/warehouse
...

• core-site.xml 

Hadoop의 core-site.xml 파일에 username 변경 후 아래 내용 추가

#core-site.xml
<property>
     <name>hadoop.proxyuser.username.hosts</name>
     <value>*</value>
  </property>
  <property>
     <name>hadoop.proxyuser.username.groups</name>
     <value>*</value>
  </property>

• hive-site.xml

Hive의 hive-site.xml 파일에 아래 내용 추가

# hive-site.xml
<property>
  <name>hive.server2.enable.doAs</name>
  <value>true</value>
</property>
<property>
  <name>hive.server2.authentication</name>
  <value>NONE</value>
</property>

• hiveserver 2 실행

$ hiveserver2
2023-06-12 16:59:58: Starting HiveServer2
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in ...
SLF4J: Found binding in ...
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
Hive Session ID = ...

---

2. HiveServer2Hook 

2-1. Connection 설정

• connection 설정

 Admin - Connections에서 새로운 Connection을 생성하고 아래와 같이 Connection Type, Host, Login(username), Port 작성 후 Extra에 아래 내용 작성

{
  "authMechanism": "LDAP"
}

connection 설정

 

2-2. Connection Test

해당 페이지 하단에 Test 버튼을 클릭하여 연결 여부를 확인

• Error

Could not connect to any of [('127.0.0.1', 10000)]

 

해당 오류의 경우, hiveserver2 실행

 

errorMessage='Failed to open new session: java.lang.RuntimeException: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.authorize.AuthorizationException): User: 사용자명 is not allowed to impersonate hive'), serverProtocolVersion=9, sessionHandle=None, configuration=None)

해당 오류의 경우, 위에서 언급한  core-site.xml 파일 설정 확인

• Success

Connection successfully tested

 

test 결과 상단에 아래와 같이 뜨면 연결 성공 👍

 

2-3. 활용 예제

사용 방법은 공식문서를 참고

 

📌 HiveServer2Hook을 사용한 간단한 예시

from airflow import DAG
from datetime import datetime
from airflow.providers.apache.hive.hooks.hive import *
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 6, 1),
    'retries': 0,
}

test_dag = DAG(
    'hive_test',
    default_args=default_args,
    schedule_interval="* */10 * * *"
)

def hive_test():
    hql = "SELECT * FROM raw_seoul LIMIT 10"
    hm = HiveServer2Hook(hiveserver2_conn_id = 'Hiveserver2_test') #connection id
    result = hm.get_records(hql) #table의 records 가져오기
    
    print(result) # 출력
            
t1 = PythonOperator(
    task_id = 'HiveServer2Hook_test',
    python_callable=simple_query,
    dag=test_dag,
)

---

3. HiveCliHook 

3-1. Connection 설정

• connection 설정

 Admin - Connections에서 새로운 Connection을 생성하고 아래와 같이 Connection Type, Host, Login(username), Port 작성 후 Extra에 아래 내용 작성

{
  "use_beeline": true,
  "auth": ""
}

connection 설정

 

3-2. 활용예제

사용방법은 공식문서를 참고

from airflow import DAG
from datetime import datetime
from airflow.providers.apache.hive.hooks.hive import *
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 6, 1),
    'retries': 0,
}

test_dag = DAG(
    'hive_test',
    default_args=default_args,
    schedule_interval="* */10 * * *"
)

def hive_test():

    arr = [[1,2],[3,4]]
    df = pd.DataFrame(arr,columns=['a','b']) 
    
    print(df)
    hh = HiveCliHook(hive_cli_conn_id='hive_cli_connect') #connection id
    hh.load_df(df=df,table='test',
               field_dict={
                            'a':'INT',
                            'b':'INT'
                        }) #데이터 프레임을 테이블로 생성하기
    
t1 = PythonOperator(task_id='HiveCliHook_test',
                    python_callable=hive_test,
                    dag=test_dag)