코드 설명

[lsiwh37249]

Airflow + Spark 코드

spark 관련 코드 https://github.com/encore-PR4/chat-etl/blob/main/src/chat_etl/spark_stream.py

airflow 관련 코드 https://github.com/encore-PR4/dags/blob/main/preprocessing.py

streamlit 관련 코드 https://github.com/encore-PR4/chat-etl/blob/main/src/chat_etl/sl/ad_page.py

[lsiwh37249]

spark 관련 코드

강조할 점 1

• pyspark 인터프리터 내에 stark streaming api를 사용하여 직접 Kafka Topic에서 데이터 습득
• 좋은점 : Consumer 코드를 작성할 필요가 없어 코드 간결성이 높아짐, Spark dataframe을 사용할 수 있어 빠른 처리가 가능해짐,
• 좋은점2 : 실시간 처리를 위한 코드 변경 용이 spark.read -> spark.readStream을 통해 배치에서 스트리밍으로 변경(단 추가적인 코드 작성 필요)

강조할 점 2

• parquet 파일 형태로 저장 데이터 파티셔닝을 통해 검색 속도 향상 /api/v1/product/ 경로 + 사용자 ID를 결합하여 파티셔닝 진행 추후 데이터가 많아질 경우 ID 기반으로 검색하여 데이터 검색 속도를 향상시킬 수 있도록 개발

  .
  └── api
  └── v1
      └── product
          ├── _SUCCESS
          ├── specific_path=16
          │   └── part-00000-522282c4-cfb9-480f-83c6-e745d35cd7c5.c000.snappy.parquet
          ├── specific_path=17
          │   └── part-00000-522282c4-cfb9-480f-83c6-e745d35cd7c5.c000.snappy.parquet
          ├── specific_path=22
          │   └── part-00000-522282c4-cfb9-480f-83c6-e745d35cd7c5.c000.snappy.parquet
          ├── specific_path=23
          │   └── part-00000-522282c4-cfb9-480f-83c6-e745d35cd7c5.c000.snappy.parquet
          ├── specific_path=24
          │   └── part-00000-522282c4-cfb9-480f-83c6-e745d35cd7c5.c000.snappy.parquet
          ├── specific_path=28
          │   └── part-00000-522282c4-cfb9-480f-83c6-e745d35cd7c5.c000.snappy.parquet
          ├── specific_path=3
          │   └── part-00000-522282c4-cfb9-480f-83c6-e745d35cd7c5.c000.snappy.parquet
          ├── specific_path=4
          │   └── part-00000-522282c4-cfb9-480f-83c6-e745d35cd7c5.c000.snappy.parquet
          ├── specific_path=50
          │   └── part-00000-522282c4-cfb9-480f-83c6-e745d35cd7c5.c000.snappy.parquet

[lsiwh37249]

airflow 관련 코드

• 3개의 task로 구성, 핵심 task는 kafka_c로 spark_submit하는 용도로 사용
• bashOperator에 ds_nodash를 사용하여 데이터 저장 시 날짜를 포함시킴 -> 저장 및 검색이 용이하도록 저장

  save_dir = f"/home/ubuntu/data/tmp/{ds_nodash}/{common_part}"  # 저장할 디렉토리 경로

[lsiwh37249]

Streamlit

• st.cache()는 써 보긴 했는데 성능은 모르겠음
• 이 파트는 사진 or 시연 때 보여주면 될 듯

[lsiwh37249]

kafka-clients-3.5.2.jar spark-sql-kafka-0-10_2.12-3.5.1.jar spark-sql-kafka-0-10_2.12-3.5.2.jar spark-streaming-kafka-0-10_2.12-3.3.4.jar