[Airflow] Airflow & Spark 연동해서 활용하기 (1)

 

 

 

 

 

이번 포스팅에선 Airlfow와 Spark의 연동에 대해 다루겠습니다.

Spark에서의 작업을  Airflow를 통해 자동화시키는 작업까지 해보겠습니다.

당연히 Airlflow와 pyspark 환경이 세팅되어 있어야 합니다.

 

 

 

* 본 포스팅은 해당 강의를 참고한 것임을 밝힙니다.

 

 

 

 

[pyspark 세팅하러 가기]

https://mengu.tistory.com/25?category=932924 

[Spark] 초기 환경 세팅 ft. 호환 문제 해결

 

 

 

목차

📃 Airflow x Spark 연동 원리

📃 연동하기

📃 mini project

 

 

 

 

Airflow x Spark 연동 원리

 

📌 원리

 

본래 Airflow에서 간단한 데이터 작업이 가능합니다. 

하지만 대량의 데이터를 옮기고, 가공하는 작업은 Airflow에서 직접 수행하지 않는 것이 좋습니다. 어디까지나 Airflow는 워크 프로우 플랫폼이기 때문입니다. 그렇다면 어떻게 Spark 작업을 수행하도록 할까요?

 

 

 

 

Airflow 안에 있는 SparkOperator를 이용하는 것입니다.

Airflow는 자체적으로 데이터를 다루기 보단, 이미 Spark 작업이 구현되어 있는 'Task 파일'을 실행시키는 것이 효율적이라 판단했습니다. 이에 따라 DAG Task들은 거대한 기능을 수행하기보단, Spark 파일을 실행시키는 것에 초점이 맞춰집니다. 워크 플로우는 더욱 간편해지고, Task들은 각자의 파일에서 콤팩트하게 관리될 수 있어 편리합니다.

 

 

 

 

연동하기

 

📌 필요한 패키지 설치

 

airflow와 spark 연동을 위한 패키지를 깔아줍니다.

pip install apache-airflow-providers-apache-spark

 

 

 

📌 기본 파일 생성 & Import

 

(1) airflow 안에 dags 폴더에 파일을 하나 생성해줍니다.

 

 

(2) 패키지들을 airflow_spark.py 파일 내에서 임포트 해줍니다.

from datetime import datetime
from airflow import DAG
# from airflow.providers.apache.spark.operators.spark_sql import SparkSqlOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

원래 SparkSqlOperator를 통해 Spark문을 실행하는 것이 가능합니다. 하지만 이 방법은 Airflow 내에서 실행시키는 방법이므로 대용량/복잡한 데이터를 처리할 땐 알맞지 않습니다. SparkSubmitOperator를 통해 위에서 언급했던 방법으로 spark 파일을 직접 실행시키도록 하겠습니다. 

 

 

 

 

📌 Connection

 

Airflow와 Spark가 연동되기 위해선 Spark Connection을 생성해줘야 합니다. Airlfow는 지정된 Connection을 바탕으로 접속 및 기능을 수행합니다. 

 

(1) Web Server 연결

(base) mingu@DESKTOP-6EBUJPM:~/airflow$ airflow webserver
  ____________       _____________
 ____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ \_ | /| / /
___  ___ |  / _  /   _  __/ _  / / /_/ /_ |/ |/ /
 _/_/  |_/_/  /_/    /_/    /_/  \____/____/|__/
Running the Gunicorn Server with:
Workers: 4 sync
Host: 0.0.0.0:8080
Timeout: 120
Logfiles: - -
Access Logformat:
=================================================================
[2022-09-15 13:18:24 +0900] [233] [INFO] Starting gunicorn 20.1.0
[2022-09-15 13:18:25 +0900] [233] [INFO] Listening at: http://0.0.0.0:8080 (233)
[2022-09-15 13:18:25 +0900] [233] [INFO] Using worker: sync
[2022-09-15 13:18:25 +0900] [234] [INFO] Booting worker with pid: 234
[2022-09-15 13:18:25 +0900] [235] [INFO] Booting worker with pid: 235
[2022-09-15 13:18:25 +0900] [236] [INFO] Booting worker with pid: 236
[2022-09-15 13:18:25 +0900] [237] [INFO] Booting worker with pid: 237

 

 

 

(2) Connection 생성

 

 

 

(3) Connection Type를 꼭 Spark로 지정해주셔야 합니다!

 

 

 

이렇게 되면 연동은 완료되었습니다. 앞으로 DAG를 작성할 때 SparkSubmitOperator를 이용하여 Spark 파일을 실행시키기만 하면 됩니다. 자세한 이해를 위해 간단한 실습을 진행해보겠습니다. 

 

 

 

다음 포스팅에서 뵙겠습니다!