[Airflow Dag] 상품 파이프 라인

[jiminchur]

Product Pipeline

• 매일 새롭게 올라오는 무신사 내 상품들을 하루에 한번씩 수집을 해와서 데이터 클렌징 후 DB에 넣는것까지 airflow로 자동화 시킬 것이다.

• 전체적인 파이프 라인 설계

  • 소분류 카테고리 코드 정보 메인DB서버에서 가져오기
  • 위에서 가져온 정보를 토대로 그전에 가져왔던 상품들 코드와 겹치지 않는 상품코드 수집
  • 상품코드들은 먼저 DB 테이블에 담고 새로운 상품코드들 임시 저장
  • 새로운 상품코드들로 상품들 상세 정보 크롤링 후 메인DB서버에 저장
  • 임시저장했던 파일들 제거

[jiminchur]

• 3/1 Dag 코드 짜기전에 jupyter에서 test 진행
  • [파일]()
• 3/3 전처리 코드 구현중 우선 test 성공한 코드 먼저 airflow dag에 맞는 코드로 변환시작 해야할것 같음

[jiminchur]

• 3/4 Airflow Start
  • 연습용 EC2 인스턴스에서 docker airflow 이미지로 시작
  • 메인 DB 서버에서 정보가져오는 dag test 성공
• 3/5
  • 새롭게 업로드 된 상품들 코드 가져와서 메인 DB 서버에 넣는 dag 성공 (BeautifulSoup으로 진행)

[jiminchur]

• 3/6 상품 상세정보 수집
  • Selenium 설치 및 버전 맞추기 진행

[jiminchur]

issue

• 셀레니움으로 수집할려고 코드 적용하고 test중 오류

• 원인

  • webdriver에서 구글 드라이버와 크롬의 버전이 맞지 않는다는 오류
  • 가상환경에서 airflow을 사용할려고 해서 headless selenium으로 이용

    해결하기 위한 시도

• ipython에서 셀레니움 test

  • ipython test에서는 성공

• test dag 하나 만들어서 test 시도

  • test dag도 성공

[jiminchur]

• 3/7 구글 크롬하고 드라이버의 버전 맞추기 성공 셀레니움으로 수집하는거 성공
  • 데이터 전처리 한번에 진행되는 코드 작성
  • 데이터 전처리 코드 dag로 바꿔서 진행 시도

[jiminchur]

test 결과 공유

• 메인DB 서버와 연동해서 정보 출력하기

  • 성공

• 새로운 상품들의 코드 가져오기

  • 1차 실패

• issue (1)

• 원인

  • sql하고 연결이 끊어짐
  • 연결은 해결

• issue (2)

• 원인

  • 한번에 전송할수있는 크기 오류
  • 나눠서 전송하는 방법으로 해결

• 2차시도
  • 성공

• 상품 코드로 상품 상세정보 가져오기

  • 1차 시도 실패

• issue

  • 접속은 됬으나 상품정보들을 가져오질 못하는 상황

• 원인

  • 저장하는 코드 문제 수정 후 해결

[jiminchur]

• 상품 상세정보 1차 전처리 진행
• 무신사 상품들을 올릴때 올리는 판매자가 직접 정보들을 적으니깐 색깔, 계절성, 등 적어 놓은게 일정하지가 않았다 어느정도의 일관성을 갖도록 단어를 매핑하는 전처리 진행

[jiminchur]

• 전처리된 정보들 DB에 넣기

[jiminchur]

• 코드 리뷰후에 코드 최적화하기

• 피드백 1 예외 처리의 과다 사용

  • 문제 : 예외 처리(try-except)를 과도하게 사용하면 코드가 복잡해지고 가독성이 떨어진다.
  • 해결책 : 상품 정보를 추출하는 과정에서 발생할 수 있는 일반적인 예외를 미리 예측하고, 이를 처리할 수 있는 보다 효율적인 방법으로 코드 최적화

    # infor
    try:
    df['fit'].iloc[i] = str(df['infor'].iloc[i]['핏']).strip('[]').replace("'",'')
    except:
    pass
    try:
    df['texture'].iloc[i] = str(df['infor'].iloc[i]['촉감']).strip('[]').replace("'",'')
    except:
    pass
    try:
    df['elasticity'].iloc[i] = str(df['infor'].iloc[i]['신축성']).strip('[]').replace("'",'')
    except:
    pass
    try:
    df['transparency'].iloc[i] = str(df['infor'].iloc[i]['비침']).strip('[]').replace("'",'')
    except:
    pass
    try:    
    df['thickness'].iloc[i] = str(df['infor'].iloc[i]['두께']).strip('[]').replace("'",'')
    except:
    pass
    try:
    df['season'].iloc[i] = str(df['infor'].iloc[i]['계절']).strip('[]').replace("'",'')
    except:
    pass

• 피드백 2 코드 중복

  • 문제 : 비슷한 기능을 하는 코드가 여러 곳에 중복되어 있어 유지보수가 어렵고, 코드의 재사용성이 떨어진다.
  • 해결책 : 중복되는 코드를 함수로 분리하여 재사용 가능하게 만들고, 함수는 범용적으로 사용될 수 있도록 설계하고, 필요한 곳에서는 이 함수를 호출하여 사용하게 코드 수정

    for i in range(1,page + 1):
    url = f'[https://www.musinsa.com/categories/item/{code}?d_cat_cd={code}brand=&list_kind=small&sort=new&sub_sort=&page={i}&display_cnt=90&group_sale=&exclusive_yn=&sale_goods=&timesale_yn=&ex_soldout=&plusDeliveryYn=&kids=&color=&price1=&price2=&shoeSizeOption=&tags=&campaign_id=&includeKeywords=&measure=](https://www.musinsa.com/categories/item/%7Bcode%7D?d_cat_cd=%7Bcode%7Dbrand=&list_kind=small&sort=new&sub_sort=&page=%7Bi%7D&display_cnt=90&group_sale=&exclusive_yn=&sale_goods=&timesale_yn=&ex_soldout=&plusDeliveryYn=&kids=&color=&price1=&price2=&shoeSizeOption=&tags=&campaign_id=&includeKeywords=&measure=)'
    response = requests.get(url)
    soup = BeautifulSoup(response.text, 'lxml')
    if str(response).split(" ")[-1][1:4] == '403':
    time.sleep(300)
    url = f'[https://www.musinsa.com/categories/item/{code}?d_cat_cd={code}brand=&list_kind=small&sort=new&sub_sort=&page={i}&display_cnt=90&group_sale=&exclusive_yn=&sale_goods=&timesale_yn=&ex_soldout=&plusDeliveryYn=&kids=&color=&price1=&price2=&shoeSizeOption=&tags=&campaign_id=&includeKeywords=&measure=](https://www.musinsa.com/categories/item/%7Bcode%7D?d_cat_cd=%7Bcode%7Dbrand=&list_kind=small&sort=new&sub_sort=&page=%7Bi%7D&display_cnt=90&group_sale=&exclusive_yn=&sale_goods=&timesale_yn=&ex_soldout=&plusDeliveryYn=&kids=&color=&price1=&price2=&shoeSizeOption=&tags=&campaign_id=&includeKeywords=&measure=)'
    response = requests.get(url)
    soup = BeautifulSoup(response.text, 'lxml')
    if str(response).split(" ")[-1][1:4] == '403':
    time.sleep(300)
    url = f'[https://www.musinsa.com/categories/item/{code}?d_cat_cd={code}brand=&list_kind=small&sort=new&sub_sort=&page={i}&display_cnt=90&group_sale=&exclusive_yn=&sale_goods=&timesale_yn=&ex_soldout=&plusDeliveryYn=&kids=&color=&price1=&price2=&shoeSizeOption=&tags=&campaign_id=&includeKeywords=&measure=](https://www.musinsa.com/categories/item/%7Bcode%7D?d_cat_cd=%7Bcode%7Dbrand=&list_kind=small&sort=new&sub_sort=&page=%7Bi%7D&display_cnt=90&group_sale=&exclusive_yn=&sale_goods=&timesale_yn=&ex_soldout=&plusDeliveryYn=&kids=&color=&price1=&price2=&shoeSizeOption=&tags=&campaign_id=&includeKeywords=&measure=)'
    response = requests.get(url)
    soup = BeautifulSoup(response.text, 'lxml')
    if str(response).split(" ")[-1][1:4] == '403':
    break

[jiminchur]

• 하나의 dag로 합친 후 test하기
• 1차 시도
• 실패
  • 중간에 오지 않은 데이터들이 있음 크롤링을 해오는 사이트에 있거나 없거나 하는정보들이 많다 그래서 근데 새롭게 올리는 상품들에게서 없는 정보들이 공통적으로 존재 확인
• 해결하기 위한 시도
  • 거의 정보들이 없더라도 중간중간에 정보들이 있는 경우도 있음 만약 특정한 날에는 없는 경우들이 있으니까 만약 값들이 존재하지 않는다면 null값으로 채우도록 예외처리
• 2차 시도
• 성공

[jiminchur]

• 최종 성공