[Airflow] - Airflow Connection

Connection

 

커넥션이란 Airflow가 외부서비스와 통신하기 위한 URL, 사용자인증정보(ID/Password), API 키 등을 저장하는 설정 객체 즉 저장소이다. Airflow를 통해서 단순히 파이썬 함수만 실행하는게 아니라 DB연결도하고 클라우드 접속도하고 API도 쏘고 할탠데

그럴때마다 필요한 인증정보들을 Connection이라는 곳에 저장합니다.

 

Airflow의 다양한 Operator

• HttpOperator : Api를 호출해서 데이터를 받아오기위한 Operator
• SQLExecuteQueryOperator : 데이터베이스에 연결한 다음 그 데이터베이스 안에 특정 SQL 쿼리를 던진다든지 작업을 진행하기 위한 Operator
• EmailOperator : Airflow를 기반으로해서 Email을 보내는것을 자동화할때 사용하는 Operator
• S3CreateBucketOperator&BigQuertInsertJobOperator: AWS나 Azure와 같은 클라우드 자원들과 연동해서 작업해야 될때 사용하는 Operator

이러한 Operator들을 각각 사용할때 Connection이라고 하는 정보들이 필요합니다.

 

Hook과의 관계

Hook은 Connection이랑 같이 설명할때 처음에는 좀 헷갈릴수 있는 개념인데, 외부시스템과 상호작용하기위한 인터페이스를 제공하는 클래스입니다. 이말자체도 좀 애매모호한데 Connection은 주로 Hook을 통해 사용된다라고 되어있습니다.

아래 그림을 보면 DAG에서 Operator를 개발을 했습니다.그러고나서 뭔가 파이프라인 작업을 실행을 하는데,그다음에 외부서비스인 데이터베이스나 클라우드에 접근을 하려고 합니다.그러고나서 Connection만 설정이 되어 있는경우 외부서비스에 대한 정보만 저장해놓는것이고,하지만 거기에 접속해가지고 작업을 수행하기 위해서는 Python 객채로 만들어져야 되는데 그럴때 활용되는게 Hook이 되겠습니다. 즉, Hook은 Connection에 저장된 외부서비스 정보를 가져와 Python 코드에서 사용할수 있는 객체로 만들어주는 역할입니다. Hook이 생성이 되면 Python 코드들의 조합을 통해서 데이터베이스나 클라우드 스토리지와 같은 외부서비스에 접근이 가능합니다.

 

HttpOperator

 

HttpOperator를 가져오고 Hook 객체를 보여드리기 위해서 다른 Operator들도 import해서 가져왔습니다.

 

BigQueryOperator안에 들어가서 밑으로 내려보게되면 hook이라고 하는것이 보이는데,

gcp_conn_id라고 하는것을 받아오게됩니다. 

gcp_conn_id  : Airflow UI에서 설정할 Connection 정보를 받아옵니다. 이 정보를 받아와서 사용할수 있는 파이썬 객채로 만들어줍니다. 이 객체를 기반으로해서 실제로 BigQuery에 SQL 작업을 던질수 있도록 밑에 코드가 개발이되어 있습니다.

 

 

GCSListObjectsOperator는 구글 클라우드 스토리지의 특정 버킷에 어떠한 객체들 리스트가 있는지를 보여주는 오퍼레이터인데 이것도 원본을 까보게 되면

 

GCSHook이 위치해 있고 첫번째 파라미터로 gcp_conn_id를 받아오는데 이 connection id 역시 Airflow UI에서 connection  안에 정보를 넣게 되면 그 정보를 받아와서 파이썬 객체를 만들고 이 객체를 기반으로해서 그 기능을 수행할수 있도록 해주는 코드입니다.

마지막으로는 Mysqldatabase 안에 있는 테이블을 구글 클라우드 스토리지의 파일 형태로 옮겨주는 Operator입니다.

이것도 안에를 까보게 되면

이것도 동일하게 Hook을 기반으로 Connection 정보를 받아와서 객체가 만들어지면 그 객체를 기반으로 Mysql에서 구글 클라우드 스토리지로 옮겨갈수 있도록 하는 역할입니다.

 

https://airflow.apache.org/docs/apache-airflow-providers-http/stable/_modules/tests/system/http/example_http.html

tests.system.http.example_http — apache-airflow-providers-http Documentation

apacheAirflow document에서도 가져올수 있습니다.

HttpOperator 실습

HttpOperator에서 task_id는 무조건 써줘야 하는곳이고, http_conn_id가 있습니다.

http_conn_id에서 UI에서 이 이름으로 등록을 해볼 예정입니다.

endpoint도 넣어주고 method도 GET으로 설정해두었습니다.

 

아래 양조장 Api 사이트인데 Api에 관련된 JSON 리스트를 50개를 리턴해주는 API를 Airflow로 쏴보도록 하겠습니다.

 

Airflow의 Add Connection을 눌러주고  Type은 Http로 설정해두고 아까 설정해둔 http_conn_id를 connection id에 넣어줍니다

그리고 API주소를 불러서 복사해서 가져오면되는데 http부터 주소까지니는 Host이고 v1/breweries가 endpoint가 됩니다.

특정 호스트를 호출할때 ENDPOINT가 조금씩 바껴줘야 되는 상황이라면 저런 방식으로 사용하는게 맞아서 세팅을 이렇게 하였습니다.

Xcom으로가서 확인해본 결과 50개의 Json형태의 Api가 들어온걸 확인해 볼수 있습니다.

 

SqlOperator 실습

 

Mysql 서비스를 Docker 컨테이너로 한대 띄우고, 그다음에 SQLExecuteQuertOperator를 적용하는데 필요한 쿼리문을 만들어 봤습니다. insert로 3개를 넣어주고 id가 3번인거를 업데이트하는 SQL문을 만들어봤습니다.

네트워크를 명시를 한후 실행하는 모습

도커컨테이너에 네트워크 이름을 명실해주고 띄워줘야되는데 이게 왜그러는거냐면, Airflow 인프라를 구성할때 docker-compose.yaml 파일을 기반으로 구성을 했는데 도커에도 네트워크라는 개념이 있습니다. 그때 컨테이너가 어떤 네트워크에 뜨게되냐면 docker -> 기본 네트워크 사용 (bridge) 라는 네트워크에 컨테이너들이 뜨게 됩니다. 하지만 docker compose에 여러대의 컨테이너를 일괄적으로 띄우게 되면, 자체적으로 네트워크를 하나 만듭니다. 즉, bridge 네트워크가 아닌 다른 네트워크를 하나 만들어서 그 안에 컨테이너들을 띄웁니다.

ls로 확인해보는 모습

docker에서 컨테이너 생성된 모습

connection을 Mysql을 연걸해주는 모습

 

airflow에서 정상적으로 작동하고 sql문에서 3번째인 update문에서 peter로 변경되어있는걸 확인해볼수 있습니다.

 

마치며 

Connection은 UI에서 직관적으로 해볼수있기 때문에 그렇게 어려운점은 없었을거라 생각이 듭니다. 다만 이렇게 많은 Connection들에 대해서 각각의 필드들이 조금씩 다릅니다. Azure 같은 경우에는 Bolb Storage 로그인이나 Key같은 필드가 있고, AWS 같은경우에는 AWS Access Key 세트를 넣는 필드들이 있습니다. EmailOperator도 필드들이 다다릅니다. Type들을 잘 가져와서 활용하는게 처음에는 조금 까다로운 부분이 될수가 있겠습니다.