인프런 커뮤니티 질문&답변

안녕하세요. 서울시 공공데이터 포털에 있는 실시간 데이터를 바탕으로, 진행해보고 싶어서 강사님께서 짜신 코드와 조금 다르게 구성을 했는데,

패키지 에러 문제가 나타났습니다. 에러 사항은 import xmltodict error문제 입니다.

(1) 데이터 수집

https://data.seoul.go.kr/dataList/OA-21285/A/1/datasetView.do

샘플 데이터 양식: http://openapi.seoul.go.kr:8088/{API_KEY}/xml/citydata/{startnum}/{endnum}/POI{num}'

POI{num}의 경우, 해당 num은 숫자가 아닌 str형태입니다. (001 ~ 113)

(저는 여기서, 전기차 충전소와 충전기 데이터 분석을 통해, 급속 충전기 최적 입지 데이터 분석을 해보고 싶어 크롤링과 MYSQL에 데이터를 수집은 해놓았습니다. )

(강사님께 airflow를 배우며, airflow를 통해 배치 기반으로 데이터 수집을 자동화하고 싶습니다.)

[충전소 데이터]

[충전기 데이터]

---

(2) 에러 발생!

(3) 코드 개요

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.decorators import task
from airflow.providers.http.operators.http import SimpleHttpOperator
# import common.poi_func as poi_array# poi nums
import pendulum

with DAG(
    dag_id="dags_simple_http_operators",
    start_date=pendulum.datetime(2023, 8, 1, tz="Asia/Seoul"),
    catchup=False,
    schedule="0 18-20 * * *" #매일 오후 6시~8시 1시간 간격으로 실행
) as dag:
    
    """ 서울시 실시간 전기차 충전소/충전기 데이터 정보 """
    # http_conn_id = 'openapi.seoul.go.kr'
    
    #array = poi_array()
    #for poi_number in array:
    tb_electric_station_info = SimpleHttpOperator(
        task_id="tb_electric_station_info",
        http_conn_id='openapi.seoul.go.kr',
        endpoint= '{{var.value.apikey_openapi_seoul_go_kr}}/xml/citydata/1/1/광화문·덕수궁',
        method='GET',
        headers={"Content-Type": "application/xml"}
    )
        
    @task(task_id='python_1')
    def python_1(**kwargs):
        ti = kwargs['ti']
        rslt = ti.xcom_pull(task_ids='tb_electric_station_info')
        from pprint import pprint
        import xmltodict
        response_dict = xmltodict.parse(rslt)
        pprint(response_dict)
    
    tb_electric_station_info >> python_1() 

강사님께서는 json.load()를 사용했지만, 해당 실시간 데이터는 xml 형식으로만 받아올 수 있어서, 출력을 하기 위해 xmltodict 를 import 해와서, 잘 받아오는지 테스트를 해보고자 하였습니다. 하나의 페이지에 무수히 많은 태그들이 있기 때문에, 페이지는 1페이지로만 설정을 했습니다.

해당 에러를 해결하기 위해서, pip install xmltodict를 진행하였습니다.

그럼에도, 해결이 되지 않아서, git이 연동되어있는 쪽에,

pip install xmltodict로 패키지를 설치해주었습니다.

혹시 몰라서, docker compose down을 시킨 뒤,

다시 docker_compose.yaml 파일이 있는 곳에서 docker compose up을 실행하여 airflow 환경에 접속을 했는데,

ModuleNotFoundError: No module named 'xmltodict' [2023-08-15, 23:47:23 KST] {taskinstance.py:1350} INFO - Marking task as FAILED. dag_id=dags_simple_http_operators, task_id=python_1, execution_date=20230815T144715, start_date=20230815T144723, end_date=20230815T144723 [2023-08-15, 23:47:23 KST] {standard_task_runner.py:109} ERROR - Failed to execute job 206 for task python_1 (No module named 'xmltodict'; 77)

여전히 해당 모듈이 없다는 에러가 뜨네요 ..

어떻게 해결해야할까요?

 ---

[추가1]

xcom에는 그래도 데이터가 잘 받아와지는 걸 확인할 수 있었습니다.

 해당 데이터를 파싱하는 과정에서 에러가 난 것 같습니다.

[추가2]

나름대로 방법을 찾아봤는데,

https://stackoverflow.com/questions/67851351/cannot-install-additional-requirements-to-apache-airflow

docker_compose.yaml파일을 수정해봤는데도...

종료가 되네요..