[Airflow] Airflow의 Python Operator / Hive Operator 사용하기

2023.06.09 (FRI) 학습정리
#Airflow #Hive

---

1. Airflow Operator

지금까지 BashOperator, EmptyOperator 같이 기본적인 operator만 사용했으나 조금 더 고도화 하기위해 다른 것들도 사용해보고자 함💪

---

2. Python Operator

python operator는 추가 설치 없이 import 후 사용가능

 

📌 python operator를 사용한 간단한 예시

from airflow import DAG
from airflow.operators.python import PythonOperator

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

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

# 리스트 형식으로 인자 전달
def get_sum(*num):
    sum = 0 
    for n in num:
    	sum += num
        
t1 = PythonOperator(task_id='python_test1',
                    python_callable=get_sum,
                    op_args=[1,2,3],
                    dag=test_dag)
 
# 딕셔너리 형식으로 인자 전달
def get_sum2(**num):
   sum = num['first'] + num['second']
                        
 t2 = PythonOperator(task_id='python_test2',
                    python_callable=get_sum2,
                    op_args={'first':1,'second':2},
                    dag=test_dag)

---

3. Hive Operator

Hive connection에는 아래와 같은 3가지 유형이 주로 사용되는데 그 중 Hive Server2 Connection, 인증방식은 LDAP 을 기반으로 진행 - 🔗 참고문서

• Hive CLI Connection
• Hive Metastore Connection
• Hive Sever2 Connection

3-1. package 설치

hive operator는 providers package 중 apache-airflow-providers-apache-hive package 설치 必

설치는 🔗 공식문서를 참고하여 진행

• Requirements 설치

아래 표 중에 설치되지 않은 패키지가 있는지 확인 후 설치 진행

pip list #설치 패키지 확인

apache-airflow

apache-airflow-providers-common-sql

hmsclient

pandas

pyhive[hive]

sasl

thrift

• 참고) sasl 설치 오류 - error: command '/usr/bin/gcc' failed with exit code 1

....
	  In file included from sasl/saslwrapper.cpp:629:
      sasl/saslwrapper.h:22:10: fatal error: sasl/sasl.h: No such file or directory
         22 | #include <sasl/sasl.h>
            |          ^~~~~~~~~~~~~
      compilation terminated.
      error: command '/usr/bin/gcc' failed with exit code 1
      [end of output]

  note: This error originates from a subprocess, and is likely not a problem with pip.
  ERROR: Failed building wheel for sasl
  Running setup.py clean for sasl
Failed to build sasl
ERROR: Could not build wheels for sasl, which is required to install pyproject.toml-based projects

위와 같은 오류가 발생한다면 아래 설치 후 sasl 설치 재진행

sudo apt-get install libsasl2-dev

• apache-airflow-providers-apache-hive 설치

pip install apache-airflow-providers-apache-hive

3-2. connection 설정

• hiveserver2 설정

connection 생성 전 아래 글 참고하여 hiveserver2 설정 진행하기

🔗 [Airflow] HiveServer2Hook 사용하여 hive에 접근하기

• connection 생성

메뉴 - Admin - Connections에 들어가서 connection 생성하기 

Admin - Connections

아래와 같이 connection Type : Hive Client Wrapper, Host, Login(username), Password, Port 입력 후 Extra에 아래 내용 작성

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

connection 설정

 

📌 hive operator를 사용한 간단한 예시

from airflow import DAG
from airflow.providers.apache.hive.operators.hive import HiveOperator

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

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

hql= '''
    CREATE TABLE test;
    '''

h1 = HiveOperator(
    task_id='HiveOperator_test',
    hql=hql,
    hive_cli_conn_id='hive_cli_connect', #connection id 작성
    run_as_owner=True,
    dag=test_dag,
)

---

+ 자료도 몇 없고 😥 공식문서도 엄청 자세하게 나와있지는 않아서 최대한 공식문서를 참고하면서 오류를 하나하나 찾아 해결하면서 진행했더니 맞는 과정인지 모르겠지만 해결하니 속시원..😌