캐글에서 가지고 온 쇼핑몰 로그 데이터(8,9월 자료)를 병합하여 탐색한 내용

데이터 합치기

스파크 세션을 띄우고 스키마를 자동으로 설정하게 하고, 헤더가 있는 자료라는 옵션을 주어 불러왔다. 이후 union을 이용하여 합쳤다.

• union - 데이터를 합치고 중복된 행을 제거
• unionAll - 데이터를 합치고 중복된 행을 유지

printSchema를 활용하여 스키마 구조 탐색

category_code 전처리

csv파일에 category_code 칼럼이 electronics.audio.headphone과 같은 형식으로 .을 기준으로 대, 중, 소분류가 되어있었다.

이를 나눠서 탐색하고자 .을 기준으로 나누고 새로운 칼럼을 생성했다. major_category, intermediate_category, minor_category 각각 대, 중, 소 분류 칼럼으로 생성

from pyspark.sql.functions import split, col

# DataFrame의 category_code 칼럼을 대, 중, 소 분류로 분할
shop = result_df.withColumn("category_list", split(col("category_code"), "\."))

# 대분류, 중분류, 소분류 칼럼 생성
shop = shop.withColumn("major_category", col("category_list")[0])
shop = shop.withColumn("intermediate_category", col("category_list")[1])
shop = shop.withColumn("minor_category", col("category_list")[2])

# 중복되는 칼럼 제거
shop = shop.drop("category_list")
# 결과 확인
shop.show()

카테고리를 대상으로 데이터를 분석하고 싶어 카테고리가 null값인 경우를 제외하였다.

shop = shop.filter(col("category_code").isNotNull())

이후 데이터의 수를 확인해보니 아래와 같이 나왔다.

EDA

”저희 고객사별로 이번 분기 프로모션 지원 대상좀 추려보려고 하는데, 현재 카테고리 별로 전환율이랑 매출 현황좀 분석해주세요” 라는 물음을 배경으로 하여 전환율 현황 분석 리포트를 위한 탐색을 했다.

합친 데이터를 임시 뷰를 만들어 EDA 해보려 한다.

shop.createOrReplaceTempView("shop")

매출

1. 대분류의 카테고리 중 가장 많은 구매 카운트 상위 10건 출력

• 전자제품, 가구, 컴퓨터 순으로 많았다.

2. 가장 많이 팔린 품목인 전자 제품을 판 브랜드 중 상위 10개 브랜드는?

• 삼성, 애플, 샤오미 순이었다.

전환율

3. event_type 확인 view, cart, purchase로 보기, 카트 담기, 구매로 나눠져 있었다.

4. 쇼핑몰에서 뷰만 하던 이용자 중에 카트로 담은 비율 확인 4%로 그리 큰 비율이 아니었다.

5. 카트로 담은 이용자 중 구매로 넘어가는 비율 확인 40%로 상대적으로 큰 비율이었다.

6. 구매하는 경로인 뷰>카트>구매와 뷰> 구매 확인 보기만 하고 바로 구매하는 경우도 있었지만 17%밖에 되지 않았고, 카트에 담은 후에 구매로 전환되는 비율은 41%로 상대적으로 더 컸다.

차트 시각화

view to cart, cart to purchase, view to purchase 3가지 항목을 비율화하여 파이차트로 표시해 본 결과 카트에 담아 구매로 넘어가는 전환율이 압도적으로 높았다.

추론

사람들이 구매하는 과정에서 여러 제품을 많이 구경한다. 상품 비교를 많이 해보는 것으로 추정된다. 하지만 많은 사람들이 본다고 해서 카트 담기로 전환하는 비율은 크지 않다. 카트 담기로 전환된 제품들은 구매로 이어질 확률이 보다가 구매하는 것보다 크게 증가한다.

카트 담기에 대한 프로모션을 진행한다면 구매 유도하는데 더 도움이 되지 않을까 추측한다. 예) 카트 담기 시 할인 쿠폰을 제공 (기간 설정)

추후

매출과 전환율을 관련지어 구매 예측 모델을 만들어 보려한다.

openVPN을 통해 학원IP가 아니더라도 연결하여 작업을 수행할 수 있게 만들어 주었다.

Spark 클러스터 모드로 데이터 전처리 하기

from pyspark.sql import SparkSession

from pyspark.sql.functions import col
from pyspark.sql.functions import explode
from pyspark.sql.functions import count
from pyspark.sql.functions import regexp_replace
from pyspark.sql.functions import when, array_contains, coalesce, udf, StringType

df = spark.read.parquet("hdfs:/killv2_/")

df.count()

df.printSchema()을 확인해보니 아래와 같이 되어있었다.

총 322656건이 자료가 조회됐다.

df = df.withColumn("victim_weapon", regexp_replace("victim_weapon", r"(?:^Weapon|^Weap)|_C.*$", ""))
df = df.withColumn("killer_weapon", regexp_replace("killer_weapon", r"(?:^Weapon|^Weap)|_C.*$", ""))

df = df.withColumn("무기분류", 
                   when(df['killer_weapon'].isin(['SKS', 'SLR', '미니 14', 'MK12', 'MK14', 'QBU', 'VSS']), 'DMR')
                   .when(df['killer_weapon'].isin(['AWM', 'KAR98K', '링스 AMR', 'M24', '모신 나강', 'WIN94']), 'SR')
                   .when(df['killer_weapon'].isin(['토미 건', 'PP-19 비존', '마이크로 UZI', 'MP5K', 'MP9', 'P90', 'UMP45', '벡터']), 'SMG')
                   .when(df['killer_weapon'].isin(['DP-28', 'M249', 'MG3']), 'LMG')
                   .when(df['killer_weapon'].isin(['S12K', 'DBS', '012', 'S1897', 'S686', '소드 오프']), 'SG')
                   .when(df['killer_weapon'].isin(['DEAGLE', 'P18C', 'P1911', 'P92', 'R1895', 'R45', '스콜피온']), 'Pistol')
                   .when(df['killer_weapon'].isin(['석궁']), 'MISC')
                   .when(df['killer_weapon'].isin(['M416', 'G36C', 'ACE32', 'AKM', 'AUG', 'FAMAS', '그로자', 'K2', 'M16A4', '베릴 M762', 'MK47 뮤턴트', 'QBZ', 'SCAR-L']), 'AR')
                   .otherwise("Unknown")
                  )


 weapon_name_mapping = {
        'SKS': 'SKS', 'FNFal': 'SLR', 'Mini14': '미니14', 'Mk12': 'MK12', 'Mk14': 'MK14',
        'QBU88': 'QBU', 'VSS': 'VSS', 'AWM': 'AWM', 'Kar98k': 'KAR98K', 'L6': '링스 AMR',
        'M24': 'M24', 'Mosin': '모신 나강', 'Win1894': 'WIN94', 'Thompson': '토미 건',
        'BizonPP19': 'PP-19 비존', 'UZI': '마이크로 UZI', 'MP5K': 'MP5K', 'MP9': 'MP9',
        'P90': 'P90', 'UMP': 'UMP45', 'Vector': '벡터', 'DP28': 'DP-28', 'M249': 'M249',
        'MG3': 'MG3', 'Saiga12': 'S12K', 'DP12': 'DBS', 'OriginS12': 'O12',
        'Winchester': 'S1897', 'Berreta686': 'S686', 'Sawnoff': '소드 오프',
        'DesertEagle': 'Deagle', 'G18': 'P18C', 'M1911': 'P1911', 'M9': 'P92',
        'NagantM1895': 'R1895', 'Rhino': 'R45', 'vz61Skorpion': '스콜피온',
        'Crossbow_1': '석궁', 'HK416': 'M416', 'G36C': 'G36C', 'ACE32': 'ACE32',
        'AK47': 'AKM', 'AUG': 'AUG', 'Mk47Mutant': 'MK47 뮤턴트', 'FamasG2': 'FAMAS',
        'G36C': 'G36C', 'K2': 'K2', 'M16A4': 'M16A4', 'BerylM762': '베릴 M762',
        'QBZ95': 'QBZ', 'SCAR-L': 'SCAR-L', 'Groza': '그로자'
    }

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

def map_weapon(weapon):
    return weapon_mapping.get(weapon, weapon)

map_weapon_udf = udf(map_weapon, StringType())
df_exploded = df.withColumn("killer_weapon", map_weapon_udf(df["killer_weapon"]))

매치 데이터가 많아져 전처리를 한꺼번에 수행하려 하면 로컬에서 실행할 경우에 커널이 죽어버리는 현상이 발생했는데, 클러스터를 형성하여 파케이 파일들을 다 불러모아 실행하니 커널이 죽지 않고 전처리가 잘 됐다.

• 간혹 파케이 파일의 스키마 형식이 잘못되어 오류가 발생하는 경우가 있었다. victim_weapon의 형태가 string이 기댓값인데, int32로 되어있는 경우가 있었기 때문이다. 그래서 아래와 같은 코드로 형태를 변형시켜서 적용하였다.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when
import re

# Spark 세션 생성
spark = SparkSession.builder.appName("WeaponPreprocessing").getOrCreate()

# 여러 개의 Parquet 파일을 동시에 로드
df = spark.read.parquet("hdfs:/killv2_/")

# victim_weapon 컬럼 데이터 타입 변환
df = df.withColumn("victim_weapon", col("victim_weapon").cast("string"))

# killer_weapon 컬럼 값 전처리
df = df.withColumn("killer_weapon", regexp_replace("killer_weapon", r"(?:^Weapon|^Weap)|_C.*$", ""))
df = df.withColumn("killer_weapon",
                   when(df['killer_weapon'] == 'SKS', 'SKS')
                   .when(df['killer_weapon'] == 'FNFal', 'SLR')
                   .when(df['killer_weapon'] == 'Mini14', '미니14')
                   .when(df['killer_weapon'] == 'Mk12', 'MK12')
                   .when(df['killer_weapon'] == 'Mk14', 'MK14')
                   # 다른 경우도 추가해주세요
                   .otherwise(df['killer_weapon'])
                   )

# 무기분류 컬럼 생성
df = df.withColumn("무기분류", 
                   when(df['killer_weapon'].isin(['SKS', 'SLR', '미니14', 'MK12', 'MK14']), 'DMR')
                   .when(df['killer_weapon'].isin(['AWM', 'KAR98K', '링스 AMR', 'M24', '모신 나강', 'WIN94']), 'SR')
                   .when(df['killer_weapon'].isin(['토미 건', 'PP-19 비존', '마이크로 UZI', 'MP5K', 'MP9', 'P90', 'UMP45', '벡터']), 'SMG')
                   .when(df['killer_weapon'].isin(['DP-28', 'M249', 'MG3']), 'LMG')
                   .when(df['killer_weapon'].isin(['S12K', 'DBS', '012', 'S1897', 'S686', '소드 오프']), 'SG')
                   .when(df['killer_weapon'].isin(['DEAGLE', 'P18C', 'P1911', 'P92', 'R1895', 'R45', '스콜피온']), 'Pistol')
                   .when(df['killer_weapon'].isin(['석궁']), 'MISC')
                   .when(df['killer_weapon'].isin(['M416', 'G36C', 'ACE32', 'AKM', 'AUG', 'FAMAS', '그로자', 'K2', 'M16A4', '베릴 M762', 'MK47 뮤턴트', 'QBZ', 'SCAR-L']), 'AR')
                   .otherwise("Unknown")
                  )

# 변환 결과 확인
df.show()

# 이제 df를 사용하여 원하는 분석 작업을 수행할 수 있습니다.

# Spark 세션 종료
spark.stop()

데이터 분석 과정 무기 상성과 무기 티어표

1. 무기 상성

개인전의 경우 killer_weapon과 finisher_weapon이 같아 문제가 없었지만, 다인전(듀오나 스쿼드) 의 경우 교전한 데이터인 killer_weapon과 마지막 한발만을 친(막타) finsiher_weapon에 차이가 있었다. 게임 내에서 교전 시에 승률을 알고 싶었기 때문에 마지막에 쏜 것이 아닌 교전을 하며 싸운 killer_weapon을 사용하고자 했다. 교전에 사용된 주무기들을 추출하여 교전 횟수가 일정 횟수 이상인 데이터만을 찾아 상대적 승률을 나타내고자 했다. AI유저의 데이터는 반영하지 않았고, 동일 무기에 대해서도 반영하지 않았다. 주무기 만의 교전 결과만을 반영하였다.

# 전체 대상 무기 추출
target = result_df['killer_weapon'].unique()

result = {}
for weapon in target:
    # 필터링: total_count가 20 이상인 데이터만 선택합니다.
    filtered_df = result_df[(result_df['killer_weapon'] == weapon) & (result_df['total_count'] >= 20)].sort_values('win_rate', ascending=False)

    high_weapons = filtered_df.iloc[:3] if filtered_df.shape[0] >= 3 else filtered_df
    low_weapons = filtered_df.iloc[-3:] if filtered_df.shape[0] >= 3 else filtered_df

    result[weapon] = {
        "highs": high_weapons,
        "lows": low_weapons,
    }

result_list = []

for weapon in result.keys():
    result_list.append({
        "weapon_name": weapon,
        **{f"easy_weapon_{i+1}": row["victim_weapon"] if row["victim_weapon"] else None for i, row in result[weapon]["highs"].reset_index(drop=True).iterrows()},
        **{f"easy_percent_{i+1}": row["win_rate"] if row["win_rate"] else None for i, row in result[weapon]["highs"].reset_index(drop=True).iterrows()},
        **{f"hard_weapon_{i+1}": row["victim_weapon"] if row["victim_weapon"] else None for i, row in result[weapon]["lows"].reset_index(drop=True).iterrows()},
        **{f"hard_percent_{i+1}": row["win_rate"] if row["win_rate"] else None for i, row in result[weapon]["lows"].reset_index(drop=True).iterrows()},
    })

final_df = pd.DataFrame(result_list)

final_df = final_df.fillna(0)

• 이 코드를 실행하여 적용하여 보니 문제가 생겼다. 슬라이싱을 활용해 승률로만 정리하다보니 교전한 무기가 3개 미만인 경우에 상대하기 어려운 무기와 쉬운 무기에 똑같은 무기가 들어가는 경우가 발생했다.

수정된 코드 코드를 승률 50퍼 기준으로 아예 다르게 변수에 저장하였다.

# 전체 경기 수를 계산합니다.
result_df['total_count'] = result_df['count'] + result_df['reverse_count']

# 승률을 계산합니다. (승리 횟수 / 전체 경기 횟수)
result_df['win_rate'] = result_df['count'] * 100 / result_df['total_count']

# None 값과 무기가 서로 같은 경우를 제거합니다.
result_df = result_df[(result_df['victim_weapon'].notnull()) & (result_df['killer_weapon'] != result_df['victim_weapon'])]

# 전체 대상 무기 추출
target = result_df['killer_weapon'].unique()

result = {}
for weapon in target:
    filtered_df = result_df[(result_df['killer_weapon'] == weapon) & (result_df['total_count'] >= 20)]
    filtered_df["high"] = filtered_df["win_rate"] >= 50  # 승률 50%를 기준으로 새로운 'high' 컬럼에 True 또는 False 값 저장합니다.

    high_weapons = filtered_df[filtered_df["high"]].sort_values('win_rate', ascending=False)[:3]
    low_weapons = filtered_df[~filtered_df["high"]].sort_values('win_rate', ascending=False)[-3:]

    high_weapons.drop('high', axis=1, inplace=True)
    low_weapons.drop('high', axis=1, inplace=True)

    result[weapon] = {
        "highs": high_weapons,
        "lows": low_weapons,
    }


result_list = []

for weapon in result.keys():
    highs = result[weapon]["highs"].reset_index(drop=True) if result[weapon]["highs"] is not None else pd.DataFrame(columns=['victim_weapon', 'win_rate'])
    lows = result[weapon]["lows"].reset_index(drop=True) if result[weapon]["lows"] is not None else pd.DataFrame(columns=['victim_weapon', 'win_rate'])

    result_list.append({
        "weapon_name": weapon,
        **{f"easy_weapon_{i + 1}": row["victim_weapon"] if not pd.isna(row["victim_weapon"]) else None for i, row in highs.iterrows()},
        **{f"easy_percent_{i + 1}": row["win_rate"] if not pd.isna(row["win_rate"]) else None for i, row in highs.iterrows()},
        **{f"hard_weapon_{i + 1}": row["victim_weapon"] if not pd.isna(row["victim_weapon"]) else None for i, row in lows.iterrows()},
        **{f"hard_percent_{i + 1}": row["win_rate"] if not pd.isna(row["win_rate"]) else None for i, row in lows.iterrows()},
    })

final_df = pd.DataFrame(result_list)
final_df = final_df.fillna(0)

• 결과값

  중복데이터도 사라지고, 승률이 높아야 할 easy_percent가 50 아래인 경우도 없어졌다.

최종 dag코드 read_data_task : 이전 포스트에서 저장했던 여러 매치 데이터에서 파싱한 무기데이터 불러오기 process_data_task : 무기 상성 및 무기 티어 데이터 처리 update_database_task : 처리한 데이터를 DB에 최신화

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
import os
from google.cloud import storage
import numpy as np
import pandas as pd
from io import BytesIO
import io
import pyarrow
# slack_notifications.py
from slack_notifications import SlackAlert
from airflow.models import Variable
import pymysql
from config import DB_CONFIG

KEY_PATH = "./playdata-2-1e60a2f219de.json"
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = KEY_PATH

slack_api_token = Variable.get("slack_api_token")
alert = SlackAlert('#message', slack_api_token) # 메세지를 보낼 슬랙 채널명을 파라미터로 넣어줍니다.


dag = DAG(
    dag_id="load_weapon_data",
    description="무기_분석",
    start_date=datetime(2023, 7, 1, 0, 0),
    schedule_interval='0 16 * * *',
    on_success_callback=alert.success_msg,
    on_failure_callback=alert.fail_msg
)

def _read_data_from_gcp_storage(**kwargs):
    bucket_name = "playdata2"
    file_path = "logs_weapon/"
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)

    parquet_data = []
    blobs = bucket.list_blobs(prefix=file_path)

    for blob in blobs:

        # parquet 형식의 파일인지 확인
        if blob.name.endswith(".parquet"):
            # 객체를 바이트 스트림으로 다운로드
            byte_stream = io.BytesIO(blob.download_as_bytes())

            # parquet 데이터를 pandas DataFrame으로 읽기
            df = pd.read_parquet(byte_stream)
            parquet_data.append(df)


    # 개별 DataFrame들을 하나의 DataFrame으로 합치기
    concat_data = pd.concat(parquet_data, axis=0, ignore_index=True)
    kwargs['ti'].xcom_push(key='parquet_data', value=concat_data)


def _process_weapon_data(**kwargs):
    # GCP Storage에서 데이터 읽어오기
    parquet_data = kwargs['ti'].xcom_pull(key='parquet_data')
    kv3 = pd.concat(parquet_data, axis=0, ignore_index=True)
    result_df = kv3.groupby(['killer_weapon', 'victim_weapon']).size().reset_index(name='count')
    reverse_combinations = result_df.rename(columns={'killer_weapon': 'victim_weapon', 'victim_weapon': 'killer_weapon', 'count': 'reverse_count'})
    result_df = result_df.merge(reverse_combinations, on=['killer_weapon', 'victim_weapon'], how='outer')
    result_df.fillna(0, inplace=True)

    result_df['total_count'] = result_df['count'] + result_df['reverse_count']
    result_df['win_rate'] = result_df['count'] * 100 / result_df['total_count']
    result_df = result_df[(result_df['victim_weapon'].notnull()) & (result_df['killer_weapon'] != result_df['victim_weapon'])]

    target = result_df['killer_weapon'].unique()
    result = {}

    for weapon in target:
        filtered_df = result_df[(result_df['killer_weapon'] == weapon) & (result_df['total_count'] >= 20)].sort_values('win_rate', ascending=False)
        high_weapons = filtered_df.iloc[:3] if filtered_df.shape[0] >= 3 else filtered_df
        low_weapons = filtered_df.iloc[-3:] if filtered_df.shape[0] >= 3 else filtered_df

        result[weapon] = {
            "highs": high_weapons,
            "lows": low_weapons,
        }

    result_list = []

    for weapon in result.keys():
        result_list.append({
            "weapon_name": weapon,
            **{f"easy_weapon_{i + 1}": row["victim_weapon"] if row["victim_weapon"] else None for i, row in result[weapon]["highs"].reset_index(drop=True).iterrows()},
            **{f"easy_percent_{i + 1}": row["win_rate"] if row["win_rate"] else None for i, row in result[weapon]["highs"].reset_index(drop=True).iterrows()},
            **{f"hard_weapon_{i + 1}": row["victim_weapon"] if row["victim_weapon"] else None for i, row in result[weapon]["lows"].reset_index(drop=True).iterrows()},
            **{f"hard_percent_{i + 1}": row["win_rate"] if row["win_rate"] else None for i, row in result[weapon]["lows"].reset_index(drop=True).iterrows()},
        })

    final_df = pd.DataFrame(result_list)
    final_df = final_df.fillna(0)

    games_threshold = 30
    weapon_summary2 = result_df.groupby('killer_weapon').agg({'total_count': 'sum', 'count': 'sum'}).reset_index()
    weapon_summary2['win_rate'] = weapon_summary2['count'] * 100 / weapon_summary2['total_count']
    valid_weapons = weapon_summary2[weapon_summary2['total_count'] >= games_threshold]

    count_weight = 0.2
    win_rate_weight = 0.8
    weapon_summary2['score'] = (weapon_summary2['total_count'] * count_weight) + (weapon_summary2['win_rate'] * win_rate_weight)

    quantiles = weapon_summary2['score'].quantile([.9, .8, .5, .3, 0]).values
    tier_1 = weapon_summary2[weapon_summary2['score'] >= quantiles[0]]
    tier_2 = weapon_summary2[(weapon_summary2['score'] >= quantiles[1]) & (weapon_summary2['score'] < quantiles[0])]
    tier_3 = weapon_summary2[(weapon_summary2['score'] >= quantiles[2]) & (weapon_summary2['score'] < quantiles[1])]
    tier_4 = weapon_summary2[(weapon_summary2['score'] >= quantiles[3]) & (weapon_summary2['score'] < quantiles[2])]
    tier_5 = weapon_summary2[weapon_summary2['score'] < quantiles[3]]

    tier_1_sorted = tier_1.sort_values('score', ascending=True)['killer_weapon'].tolist()
    tier_2_sorted = tier_2.sort_values('score', ascending=True)['killer_weapon'].tolist()
    tier_3_sorted = tier_3.sort_values('score', ascending=True)['killer_weapon'].tolist()
    tier_4_sorted = tier_4.sort_values('score', ascending=True)['killer_weapon'].tolist()
    tier_5_sorted = tier_5.sort_values('score', ascending=True)['killer_weapon'].tolist()

    for idx, row in weapon_summary2.iterrows():
        if row['killer_weapon'] in tier_1_sorted:
            weapon_summary2.loc[idx, 'tier'] = 1
        elif row['killer_weapon'] in tier_2_sorted:
            weapon_summary2.loc[idx, 'tier'] = 2
        elif row['killer_weapon'] in tier_3_sorted:
            weapon_summary2.loc[idx, 'tier'] = 3
        elif row['killer_weapon'] in tier_4_sorted:
            weapon_summary2.loc[idx, 'tier'] = 4
        else:
            weapon_summary2.loc[idx, 'tier'] = 5

    weapon_summary2['tier'] = weapon_summary2['tier'].astype(int)

    final_df = final_df.merge(weapon_summary2[['killer_weapon', 'tier']], left_on='weapon_name', right_on='killer_weapon', how='left').drop(columns=['killer_weapon'])

    output_filename = "result.csv"
    final_df.to_csv(output_filename, index=False)

    return final_df

def _update_database(final_df):
    conn = pymysql.connect(**DB_CONFIG)
    cur = conn.cursor()

    for index, row in final_df.iterrows():
        weapon_name = row['weapon_name']
        easy_weapon_1 = row['easy_weapon_1']
        easy_weapon_2 = row['easy_weapon_2']
        easy_weapon_3 = row['easy_weapon_3']
        easy_percent_1 = row['easy_percent_1']
        easy_percent_2 = row['easy_percent_2']
        easy_percent_3 = row['easy_percent_3']
        hard_weapon_1 = row['hard_weapon_1']
        hard_weapon_2 = row['hard_weapon_2']
        hard_weapon_3 = row['hard_weapon_3']
        hard_percent_1 = row['hard_percent_1']
        hard_percent_2 = row['hard_percent_2']
        hard_percent_3 = row['hard_percent_3']
        weapon_tier = row['tier']

    update_query = """
        UPDATE services_weapons SET weapon_tier = %s,
                            first_easy_weapon = %s,
                            first_easy_percent = %s,
                            second_easy_weapon = %s,
                            second_easy_percent = %s,
                            third_easy_weapon = %s,
                            third_easy_percent = %s,
                            first_hard_weapon = %s,
                            first_hard_percent = %s,
                            second_hard_weapon = %s,
                            second_hard_percent= %s,
                            third_hard_weapon = %s,
                            third_hard_percent = %s WHERE weapon_name = %s

        """
    cur.execute(update_query, (weapon_tier, easy_weapon_1, easy_percent_1, easy_weapon_2, easy_percent_2, easy_weapon_3, easy_percent_3,
                                hard_weapon_1, hard_percent_1, hard_weapon_2, hard_percent_2, hard_weapon_3, hard_percent_3,
                                weapon_name))

    conn.commit()    


read_data_task = PythonOperator(
    task_id='read_data',
    python_callable=_read_data_from_gcp_storage,
    provide_context=True
)

process_data_task = PythonOperator(
    task_id='process_data',
    python_callable=_process_weapon_data,
    provide_context=True
)

update_database_task = PythonOperator(
    task_id='update_database',
    python_callable=_update_database,
    provide_context=True
)

read_data_task >> process_data_task >> update_database_task

회고 배틀그라운드라는 게임을 많이 해보지 않았지만, 이번 프로젝트를 수행하며 배틀그라운드 게임도 많이 해보고, 특히 무기 관련해서 많이 보게 되었다. 처음에는 어느 총이 좋은 총인지도 몰라 해메었지만, 이번 분석을 통해 어느 정도 티어가 확립되고 나니 나에게 맞는 총, 그 총에 대한 정보를 알 수 있었다. 아쉬운점 아쉬운 점이 있다면, 주무기만을 한정해서 만든게 좀 아쉬웠다. 데이터를 파싱하는 데에 시간이 많이 들어 범위를 축소해서 분석을 수행했었다. 시간이 더 있었다면 다른 종류에 무기들에 대해서도 조사해보고 싶다. ( 투척무기, 근접무기, 권총 등)

나는 이 프로젝트에서 무기 데이터 부분을 맡았다. RAW DATA에서 무기 데이터 관련만 파싱하는 코드, 그 데이터를 가지고 무기 티어, 무기 별 상성을 최신화 하는 airflow dag 코드를 작성했다.

완성 화면

과정

앞서서 유저 정보 페이지에서 유저가 가장 많이 사용하는 숙련도 top3을 보여주어 자신이 가장 잘 다룰 수 있는 무기를 보여주었다. 무기들 중 주무기(소총, 저격총 등등)을 대상으로 자기가 잘 다루는 무기가 어떤 무기에는 유리하고, 또 어떤 무기에는 불리한 지 알고 싶어졌다. 1주간에 매치 데이터를 모아 이 데이터를 기반으로 무기별 상성에 대해 알아보고자 했다. 또 이를 보기 쉽게 티어표로 나타내고자 하였다.

데이터 확인 후 작업 과정

여러개의 매치 데이터가 모아 파케이 파일 형태로 저장하였다. 그 저장된 데이터를 불러모아 칼럼들을 살펴보았다.

• 로그데이터에 _T 칼럼에서 LogPlayerKillV2인 것에서 주요 칼럼으로 victim_weapon, killer_weapon, finisher_weapon을 찾았다. 개인전의 경우 killer_weapon과 finisher_weapon이 같아 문제가 없었지만, 듀오나 스쿼드의 경우 교전한 데이터인 killer_weapon과 마지막 한발만을 친(막타) finsiher_weapon에 차이가 있었다. 무기 상성에서는 victim_weapon과 killer_weapon을 사용하기로 했다.

• 로그 데이터에서 아까 가져오기로 한 데이터들만을 가져오는 코드를 만들었다. ai를 제외하고 유저의 데이터만을 가져왔다.

def _process_logs_weapon(combined_data):
tmp = [data for data2 in combined_data['logs'] for data in data2 if '_T' in data and data['_T'] == 'LogPlayerKillV2']

kv2 = []

for log in tmp:

    # 무기 관련 데이터 파싱
    if log['_T'] == 'LogPlayerKillV2':
        try :
            if 'ai' not in log['victim']['accountId'] and 'ai' not in log['killer']['accountId'] and 'ai' not in log['finisher']['accountId'] :
                try :
                    v2row = {'victim_weapon' : None if len(log['victimWeapon']) == 0 else log['victimWeapon'],
                            'victim_account_id' :log['victim']['accountId'],
                            'victim_parts' : None if len(log["victimWeaponAdditionalInfo"]) == 0 else log["victimWeaponAdditionalInfo"],
                            'killer_weapon' : log['killerDamageInfo']['damageCauserName'],
                            'killer_account_id' : log['killer']['accountId'],
                            'killer_parts' : None if len(log['killerDamageInfo']['additionalInfo']) == 0 else log['killerDamageInfo']['additionalInfo'],
                            'killer_distance' :log['killerDamageInfo']['distance'],
                            'finisher_weapon' : log['finishDamageInfo']['damageCauserName'],
                            'finisher_account_id' : log['finisher']['accountId'],
                            'finisher_parts' : None if len(log['finishDamageInfo']['additionalInfo']) == 0 else log['finishDamageInfo']['additionalInfo'],
                            'finisher_distance' : log['finishDamageInfo']['distance'],
                            }
                except :
                    v2row = {'victim_weapon' : None,
                            'victim_account_id' : None,
                            'victim_parts' : None,
                            'killer_weapon' : None,
                            'killer_account_id' : None,
                            'killer_parts' : None,
                            'killer_distance' : None,
                            'finisher_weapon' : None,
                            'finisher_account_id' : None,
                            'finisher_parts' : None,
                            'finisher_distance' : None,
                            }
            kv2.append(v2row)
        except :
            pass

kv3 = pd.DataFrame(kv2)
return kv3

• 결과값

  Weapon data에 전처리가 필요해보였다. 라이선스 만료로 인해 변수명이 실제 게임 내에 사용되는 무기명과 다른 경우가 있었고, 'weap' 'C2'등의 파츠 구분으로 인해 같은 총기 구분을 해줘야 할 필요성을 느꼈다.

• 전처리 코드

    def remove_part(text):
        return re.sub(r'(?:^Weapon|^Weap)|_C.*$', '', text)
  
    kv3['victim_weapon'] = kv3['victim_weapon'].astype(str)
    kv3['victim_weapon'] = kv3['victim_weapon'].apply(remove_part)
    kv3['killer_weapon'] = kv3['killer_weapon'].astype(str)
    kv3['killer_weapon'] = kv3['killer_weapon'].apply(remove_part)
    kv3['finisher_weapon'] = kv3['finisher_weapon'].astype(str)
    kv3['finisher_weapon'] = kv3['finisher_weapon'].apply(remove_part)
  
    weapon_name_mapping = {
        'SKS': 'SKS', 'FNFal': 'SLR', 'Mini14': '미니14', 'Mk12': 'MK12', 'Mk14': 'MK14',
        'QBU88': 'QBU', 'VSS': 'VSS', 'AWM': 'AWM', 'Kar98k': 'KAR98K', 'L6': '링스 AMR',
        'M24': 'M24', 'Mosin': '모신 나강', 'Win1894': 'WIN94', 'Thompson': '토미 건',
        'BizonPP19': 'PP-19 비존', 'UZI': '마이크로 UZI', 'MP5K': 'MP5K', 'MP9': 'MP9',
        'P90': 'P90', 'UMP': 'UMP45', 'Vector': '벡터', 'DP28': 'DP-28', 'M249': 'M249',
        'MG3': 'MG3', 'Saiga12': 'S12K', 'DP12': 'DBS', 'OriginS12': 'O12',
        'Winchester': 'S1897', 'Berreta686': 'S686', 'Sawnoff': '소드 오프',
        'DesertEagle': 'Deagle', 'G18': 'P18C', 'M1911': 'P1911', 'M9': 'P92',
        'NagantM1895': 'R1895', 'Rhino': 'R45', 'vz61Skorpion': '스콜피온',
        'Crossbow_1': '석궁', 'HK416': 'M416', 'G36C': 'G36C', 'ACE32': 'ACE32',
        'AK47': 'AKM', 'AUG': 'AUG', 'Mk47Mutant': 'MK47 뮤턴트', 'FamasG2': 'FAMAS',
        'G36C': 'G36C', 'K2': 'K2', 'M16A4': 'M16A4', 'BerylM762': '베릴 M762',
        'QBZ95': 'QBZ', 'SCAR-L': 'SCAR-L', 'Groza': '그로자'
    }
  
    # 무기 이름 변환
    kv3['victim_weapon'] = kv3['victim_weapon'].map(weapon_name_mapping)
    kv3['killer_weapon'] = kv3['killer_weapon'].map(weapon_name_mapping)
    kv3['finisher_weapon'] = kv3['finisher_weapon'].map(weapon_name_mapping)
  
    # 무기 분류를 위한 조건과 값 설정
    conditions = [
        kv3['killer_weapon'].isin(['SKS', 'SLR', '미니14', 'MK12', 'MK14', 'QBU', 'VSS']),
        kv3['killer_weapon'].isin(['AWM', 'KAR98K', '링스 AMR', 'M24', '모신 나강', 'WIN94']),
        kv3['killer_weapon'].isin(['토미 건', 'PP-19 비존', '마이크로 UZI', 'MP5K', 'MP9', 'P90', 'UMP45', '벡터']),
        kv3['killer_weapon'].isin(['DP-28', 'M249', 'MG3']),
        kv3['killer_weapon'].isin(['S12K', 'DBS', 'O12', 'S1897', 'S686', '소드 오프']),
        kv3['killer_weapon'].isin(['Deagle', 'P18C', 'P1911', 'P92', 'R1895', 'R45', '스콜피온']),
        kv3['killer_weapon'].isin(['석궁']),
        kv3['killer_weapon'].isin(['M416', 'G36C', 'ACE32', 'AKM', 'AUG', 'FAMAS', '그로자', 'K2', 'M16A4', '베릴 M762', 'MK47 뮤턴트', 'QBZ', 'SCAR-L'])
    ]
  
    values = ['DMR', 'SR', 'SMG', 'LMG', 'SG', 'PISTOL', 'MISC', 'AR']
  
    kv3['weapon_classification'] = np.select(conditions, values)
    kv3.reset_index(drop=True, inplace=True)

• 전처리 후 모습

  

dag코드 read_data_task : Raw 데이터 불러오기 process_logs_weapon_task : 필요한 무기 데이터 파싱 및 전처리 upload_data_task : 전처리 한 결과를 저장 dag가 잘도는 지 확인하기 위해 slack을 통해 알림을 받았다. 슬랙알림보내기 << 링크 확인

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
import os
from google.cloud import storage
import re
import numpy as np
import pandas as pd
import json
from io import BytesIO
import io
import pyarrow
# slack_notifications.py
from slack_notifications import SlackAlert
from airflow.models import Variable

KEY_PATH = "./playdata-2-1e60a2f219de.json"
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = KEY_PATH

slack_api_token = Variable.get("slack_api_token")
alert = SlackAlert('#message', slack_api_token) # 메세지를 보낼 슬랙 채널명을 파라미터로 넣어줍니다.


dag = DAG(
    dag_id="parsing_weapon_data",
    description="무기_분석",
    start_date=datetime(2023, 7, 2, 0, 0),
    schedule_interval='0 14 * * *',
    on_success_callback=alert.success_msg,
    on_failure_callback=alert.fail_msg
)

def _read_data_from_gcp_storage():
    bucket_name = "playdata2"
    file_path = ""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)

    parquet_data = []
    blobs = bucket.list_blobs(prefix=file_path)

    for blob in blobs:

        # parquet 형식의 파일인지 확인
        if blob.name.endswith(".parquet"):
            # 객체를 바이트 스트림으로 다운로드
            byte_stream = io.BytesIO(blob.download_as_bytes())

            # parquet 데이터를 pandas DataFrame으로 읽기
            df = pd.read_parquet(byte_stream)
            parquet_data.append(df)


    # 개별 DataFrame들을 하나의 DataFrame으로 합치기
    concat_data = pd.concat(parquet_data, axis=0, ignore_index=True)
    return concat_data



def _process_logs_weapon(**kwargs):
    ti = kwargs['ti']
    concat_data = ti.xcom_pull(task_ids='read_data')

    tmp = [data for data2 in concat_data['logs'] for data in data2 if '_T' in data and data['_T'] == 'LogPlayerKillV2']

    kv2 = []

    for log in tmp:

        # 무기 관련 데이터 파싱
        if log['_T'] == 'LogPlayerKillV2':
            try :
                if 'ai' not in log['victim']['accountId'] and 'ai' not in log['killer']['accountId'] and 'ai' not in log['finisher']['accountId'] :
                    try :
                        v2row = {'victim_weapon' : None if len(log['victimWeapon']) == 0 else log['victimWeapon'],
                                'victim_account_id' :log['victim']['accountId'],
                                'victim_parts' : None if len(log["victimWeaponAdditionalInfo"]) == 0 else log["victimWeaponAdditionalInfo"],
                                'killer_weapon' : log['killerDamageInfo']['damageCauserName'],
                                'killer_account_id' : log['killer']['accountId'],
                                'killer_parts' : None if len(log['killerDamageInfo']['additionalInfo']) == 0 else log['killerDamageInfo']['additionalInfo'],
                                'killer_distance' :log['killerDamageInfo']['distance'],
                                'finisher_weapon' : log['finishDamageInfo']['damageCauserName'],
                                'finisher_account_id' : log['finisher']['accountId'],
                                'finisher_parts' : None if len(log['finishDamageInfo']['additionalInfo']) == 0 else log['finishDamageInfo']['additionalInfo'],
                                'finisher_distance' : log['finishDamageInfo']['distance'],
                                }
                    except :
                        v2row = {'victim_weapon' : None,
                                'victim_account_id' : None,
                                'victim_parts' : None,
                                'killer_weapon' : None,
                                'killer_account_id' : None,
                                'killer_parts' : None,
                                'killer_distance' : None,
                                'finisher_weapon' : None,
                                'finisher_account_id' : None,
                                'finisher_parts' : None,
                                'finisher_distance' : None,
                                }
                kv2.append(v2row)
            except :
                pass

    kv3 = pd.DataFrame(kv2)



    def remove_part(text):
        return re.sub(r'(?:^Weapon|^Weap)|_C.*$', '', text)

    kv3['victim_weapon'] = kv3['victim_weapon'].astype(str)
    kv3['victim_weapon'] = kv3['victim_weapon'].apply(remove_part)
    kv3['killer_weapon'] = kv3['killer_weapon'].astype(str)
    kv3['killer_weapon'] = kv3['killer_weapon'].apply(remove_part)
    kv3['finisher_weapon'] = kv3['finisher_weapon'].astype(str)
    kv3['finisher_weapon'] = kv3['finisher_weapon'].apply(remove_part)

    weapon_name_mapping = {
        'SKS': 'SKS', 'FNFal': 'SLR', 'Mini14': '미니14', 'Mk12': 'MK12', 'Mk14': 'MK14',
        'QBU88': 'QBU', 'VSS': 'VSS', 'AWM': 'AWM', 'Kar98k': 'KAR98K', 'L6': '링스 AMR',
        'M24': 'M24', 'Mosin': '모신 나강', 'Win1894': 'WIN94', 'Thompson': '토미 건',
        'BizonPP19': 'PP-19 비존', 'UZI': '마이크로 UZI', 'MP5K': 'MP5K', 'MP9': 'MP9',
        'P90': 'P90', 'UMP': 'UMP45', 'Vector': '벡터', 'DP28': 'DP-28', 'M249': 'M249',
        'MG3': 'MG3', 'Saiga12': 'S12K', 'DP12': 'DBS', 'OriginS12': 'O12',
        'Winchester': 'S1897', 'Berreta686': 'S686', 'Sawnoff': '소드 오프',
        'DesertEagle': 'Deagle', 'G18': 'P18C', 'M1911': 'P1911', 'M9': 'P92',
        'NagantM1895': 'R1895', 'Rhino': 'R45', 'vz61Skorpion': '스콜피온',
        'Crossbow_1': '석궁', 'HK416': 'M416', 'G36C': 'G36C', 'ACE32': 'ACE32',
        'AK47': 'AKM', 'AUG': 'AUG', 'Mk47Mutant': 'MK47 뮤턴트', 'FamasG2': 'FAMAS',
        'G36C': 'G36C', 'K2': 'K2', 'M16A4': 'M16A4', 'BerylM762': '베릴 M762',
        'QBZ95': 'QBZ', 'SCAR-L': 'SCAR-L', 'Groza': '그로자'
    }

    # 무기 이름 변환
    kv3['victim_weapon'] = kv3['victim_weapon'].map(weapon_name_mapping)
    kv3['killer_weapon'] = kv3['killer_weapon'].map(weapon_name_mapping)
    kv3['finisher_weapon'] = kv3['finisher_weapon'].map(weapon_name_mapping)

    # 무기 분류를 위한 조건과 값 설정
    conditions = [
        kv3['killer_weapon'].isin(['SKS', 'SLR', '미니14', 'MK12', 'MK14', 'QBU', 'VSS']),
        kv3['killer_weapon'].isin(['AWM', 'KAR98K', '링스 AMR', 'M24', '모신 나강', 'WIN94']),
        kv3['killer_weapon'].isin(['토미 건', 'PP-19 비존', '마이크로 UZI', 'MP5K', 'MP9', 'P90', 'UMP45', '벡터']),
        kv3['killer_weapon'].isin(['DP-28', 'M249', 'MG3']),
        kv3['killer_weapon'].isin(['S12K', 'DBS', 'O12', 'S1897', 'S686', '소드 오프']),
        kv3['killer_weapon'].isin(['Deagle', 'P18C', 'P1911', 'P92', 'R1895', 'R45', '스콜피온']),
        kv3['killer_weapon'].isin(['석궁']),
        kv3['killer_weapon'].isin(['M416', 'G36C', 'ACE32', 'AKM', 'AUG', 'FAMAS', '그로자', 'K2', 'M16A4', '베릴 M762', 'MK47 뮤턴트', 'QBZ', 'SCAR-L'])
    ]

    values = ['DMR', 'SR', 'SMG', 'LMG', 'SG', 'PISTOL', 'MISC', 'AR']

    kv3['weapon_classification'] = np.select(conditions, values)
    kv3.reset_index(drop=True, inplace=True)

    return kv3


def _upload_data_to_gcp_storage(**kwargs):
    ti = kwargs['ti']
    df = ti.xcom_pull(task_ids='process_logs_weapon')
    storage_client = storage.Client()

    bucket_name = "playdata2"
    folder_name = "logs_weapon"

    date = datetime.today().strftime("%Y%m%d")

    buffer = BytesIO()
    df.to_parquet(buffer, engine='pyarrow', index=False)
    buffer.seek(0)

    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(f"{folder_name}/{date}_logs_weapon.parquet")
    blob.upload_from_file(buffer)

    buffer.close()

read_data_task = PythonOperator(
    task_id='read_data',
    python_callable=_read_data_from_gcp_storage,
    provide_context=True,
    dag=dag,
)

process_logs_weapon_task = PythonOperator(
    task_id='process_logs_weapon',
    python_callable=_process_logs_weapon,
    provide_context=True,
    dag=dag,
)

upload_data_task = PythonOperator(
    task_id='upload_data',
    python_callable=_upload_data_to_gcp_storage,
    provide_context=True,
    dag=dag,
)

read_data_task >> process_logs_weapon_task >> upload_data_task

7월 프로젝트로 진행하였던 배틀그라운드 API를 활용한 사이트(pd.gg)에 웹 기능 구상도 내가 주로 다루었던 페이지는 빨간색으로 표시한 무기 분석 페이지다.

1. 필요한 데이터 파싱하기

배틀그라운드 개발자 센터 << 개인당 2개의 API Key를 받을 수 있음 (1분당 10회에 요청 제한이 있음)

무기 정보 조회 데이터

유저 정보 페이지에서 본인의 무기 숙련도 Top3을 보여줄 때 사용

유저 간 전투 데이터

무기 분석 페이지와 무기 상세 페이지에서 무기 티어와 상성 간 승률을 나타내기 위해 사용

ERD

무기 마스터리 데이터 가져와서 숙련도 TOP3 보여주기

import requests

# 개인 정보 변수
api_key = 'YOUR_PUBG_API_KEY'  # 여기에 본인의 PUBG API 키를 입력하세요
player_id = 'YOUR_PLAYER_ID'   # 여기에 플레이어 ID를 입력하세요
platform = 'kakao'             # 플랫폼 정보 (예: kakao, steam 등)

def get_top3_weapon_mastery(api_key, platform, player_id):

    url = f"https://api.pubg.com/shards/{platform}/players/{player_id}/weapon_mastery"
    headers = {
        "Authorization": API_KEY,
        "Accept": "application/vnd.api+json"
    }
    response = requests.get(url, headers=headers)

    # 응답 처리
    if response.status_code == 200:
        data = response.json()
        weapon_mastery = data['data']['attributes']['weaponSummaries']

        # 경험치를 기준으로 무기 정렬
        sorted_weapons = sorted(weapon_mastery.items(), key=lambda x: x[1]['XPTotal'], reverse=True)

        # 상위 3개 무기 출력
        top3_weapons = sorted_weapons[:3]

        for idx, weapon in enumerate(top3_weapons, start=1):
            weapon_name = weapon[0].split('_')[-2] # Split the string to get the weapon name
            experience = weapon[1]['XPTotal']
            print(f"Top {idx}: {weapon_name} - 경험치: {experience}")
    else:
        print("API 요청이 실패했습니다. 상태 코드:", response.status_code)

player_id = player_id  # 플레이어 ID
api_key = api_key

get_top3_weapon_mastery(api_key, platform, player_id)

• 결과값

8대로 구성된 하둡, 스파크 클러스터를 활용하여 데이터 분석

흐름도 (예상안)

• 데이터셋 - (eCommerce behavior data from multi category store)

데이터셋 구조

데이터 로드

캐글 사이트에 접속 후 데이터를 다운로드 받았다. (위의 데이터셋 링크 참조)

데이터의 압축 해제 후 hdfs dfs -put ./2019-Nov.csv /08

HDFS 08폴더에 저장하였다.

데이터 전처리

데이터 불러오기 df = spark.read.option("header", "true").csv("hdfs:/08/2019-Nov2.csv")

스파크 세션 생성

• Raw Data.ver

• Modified.ver (전처리 후 버전 —> 카테고리 대/중/소 분할 등 처리)

  "."을 기준으로 칼럼 나누기 split_col = split(df["category_code"], "\\.") df = df.withColumn("major", split_col.getItem(0)) df = df.withColumn("intermediate", split_col.getItem(1)) df = df.withColumn("minor", split_col.getItem(2))

데이터 전처리 후 데이터 마트 생성

데이터에 category_code가 null값인 것을 제외하고 카테고리를 .을 기준으로 나눠 대, 중, 소분류 칼럼을 만들어 google bigquery에 적재하였다.

from pyspark.sql.functions import col
from google.cloud import bigquery
from google.oauth2 import service_account
import pandas as pd
import pyarrow

# category_code가 널이 아닌 레코드 필터링
filtered_df = df.filter(col("category_code").isNotNull())

# GCP 서비스 계정 키 파일 경로
key_path = "키파일경로" json 파일

# 프로젝트 및 데이터셋 ID
project_id = "프로젝트 ID"
dataset_id = "프로젝트 ID.데이터셋 ID"

# BigQuery 클라이언트 설정
client = bigquery.Client.from_service_account_json(key_path)

# 데이터 프레임을 팬더스 데이터 프레임으로 변환(데이터 시리얼라이즈)
# pandas_df = df.select("*").toPandas()
pandas_df = df.limit(100000).toPandas()
# BigQuery로 데이터 프레임 전송
table_id = "프로젝트 ID.데이터셋 ID.테이블"

job_config = bigquery.LoadJobConfig(
    # 스키마 자동 감지
    schema=[],
    autodetect=True,
    # 데이터 쓰기 방식 선택
    write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,  # 이전 테이블 데이터 대체
)

# BigQuery에 데이터 프레임 불러오기
job = client.load_table_from_dataframe(
    pandas_df, table_id, job_config=job_config
)

# 작업 완료 시간 기록
job.result()

데이터 분석

1. 카테고리 별 판매 상품 개수 집계 (TOP10)

스파크를 사용한 것과 사용하지 않은 것에 차이가 59초 차이 약 5.54배 정도에 성능을 얻을 수 있었다.

• Spark 사용

• 로컬

추후 작업 매출, 전환율 현황 분석 리포트를 카테고리 별로 나누어 보기 좋게 제작하고자 한다!

이전 포스트에서 다뤘던 spark 세션을 띄우고자 했지만, 실패한 것에 해결에 대한 기록

pyspark --master yarn --num-executors 5 명령어를 이용하여 YARN으로 PySpark를 실행하였지만, 이건 pyspark kernel이 아니라 Python 3 이었다.

단일 노드에서 spark를 사용하는 로컬 세션을 생성했던 것이었다. 어쩐지 경로 보려고 tab키를 눌렀는데, 로컬 경로가 나왔었다.

해결

크게는 2가지 pyspark 세션 생성. python 버전을 낮추는 것이었다.

해결 과정

1. Pyspark 커널 만들기

pyspark를 CLI환경이 아니라 Jupyter를 사용해서 작업 환경을 만들고 싶었기 때문에 PySpark 커널의 명령 줄 인수, 사용할 프로그래밍 언어, 추가 메타데이터, 환경 변수 등을 정의하였다.

2. shell.py 만들기

Pyspark 실행 환경을 설정하고 초기화 하는 코드를 가지고 있는 py파일

import atexit

import os

import platform

import warnings

​

from pyspark.context import SparkContext

from pyspark.sql import SparkSession

​

if os.environ.get("SPARK_EXECUTOR_URI"):

    SparkContext.setSystemProperty("spark.executor.uri", os.environ["SPARK_EXECUTOR_URI"])

​

SparkContext._ensure_initialized()  # type: ignore

​

try:

    spark = SparkSession._create_shell_session()  # type: ignore

except Exception:

    import sys

    import traceback

    warnings.warn("Failed to initialize Spark session.")

    traceback.print_exc(file=sys.stderr)

    sys.exit(1)

​

sc = spark.sparkContext

sql = spark.sql

atexit.register((lambda sc: lambda: sc.stop())(sc))

​

# for compatibility

sqlContext = spark._wrapped

sqlCtx = sqlContext

​

print(r"""Welcome to

      ____              __

     / __/__  ___ _____/ /__

    _\ \/ _ \/ _ `/ __/  '_/

   /__ / .__/\_,_/_/ /_/\_\   version %s

      /_/

""" % sc.version)

print("Using Python version %s (%s, %s)" % (

    platform.python_version(),

    platform.python_build()[0],

    platform.python_build()[1]))

print("Spark context Web UI available at %s" % (sc.uiWebUrl))

print("Spark context available as 'sc' (master = %s, app id = %s)." % (sc.master, sc.applicationId))

print("SparkSession available as 'spark'.")

​

# The ./bin/pyspark script stores the old PYTHONSTARTUP value in OLD_PYTHONSTARTUP,

# which allows us to execute the user's PYTHONSTARTUP file:

_pythonstartup = os.environ.get('OLD_PYTHONSTARTUP')

if _pythonstartup and os.path.isfile(_pythonstartup):

    with open(_pythonstartup) as f:
               code = compile(f.read(),_pythonstartup, 'exec')
               exec(code)

3. Python 버전 낮추기

💡 Tip venv는 설치된 파이썬 버전으로만 가상환경을 생성할 수 있습니다. 예를 들어, 여러분이 시스템에 Python 3.9 버전을 설치했는데 3.8 버전으로 프로젝트를 진행하고자 한다면 venv로는 이 문제를 해결할 수 없습니다. 파이썬 3.8 버전도 여러분의 PC에 설치를 해야합니다. 이와 달리 아나콘다의 conda는 아나콘다 배포판의 파이썬이 3.9 버전이라고 하더라도 3.8 버전으로 가상환경을 만들 수 있습니다.

python 3.10 버전을 사용중이었는데, Spark와 호환이 안된다는 에러 문구가 나와 가상환경을 이용하여 3.8버전으로 낮춰주었다.

conda create -n py38 python=3.8 # py38 이름을 가진 가상환경을 만들어 python 3.8버전 설치

conda env list 가상환경 생성되었나 확인

conda activate py38 py38 가상환경 접속

.bashrc 수정

export PYSPARK_DRIVER_PYTHON=jupyter
export PYSPARK_DRIVER_PYTHON_OPTS='notebook --ip=0.0.0.0'
export PKG_CONFIG_PATH="/path/to:$PKG_CONFIG_PATH"

이후에는 py38 가상환경으로 접속하여 notebook --ip=0.0.0.0 명령어를 통해 스파크를 실행하는데 성공하였다.

해결한 후 화면

배틀그라운드 데이터 하둡에 저장하기

팀원들이 개인 pc나 kict 인프라를 통해 적재해온 배틀그라운드 로그 데이터를 client pc에 모두 옮겨담았다. hdfs dfs -put ./hdfs/b/parquets/killv2/* /killv2 명령어를 사용하여 client 로컬에 있던 파일들을 모두 하둡에 저장하였다.

• 저장

(base) hadoop@client:~/hdfs/b/parquets/killv2$ ls -l | grep "^-" | wc -l

193595개의 파케이 파일을 업로드하여 데이터 준비를 마쳤다.

목표 : 프로젝트 중 무기 데이터의 집계를 하기 위한 데이터를 주기적으로 받아 GCP 스토리지에 저장해야 하는 작업이 필요했다. Raw Data를 받아와 우리가 사용할 데이터로 전처리하고 파싱하는 dag를 짰고, 이것이 잘 수행 중인지에 대해 slack bot을 통해 알림을 받고자 한다.

봇 만들기

1. https://api.slack.com/ 사이트에서 로그인 후 Your apps

2. Create New App 버튼 클릭 후 첫번째 클릭 (From scratch)

3. 앱이름을 설정 한 후 본인이 사용할 워크스페이스를 선택

4. 왼쪽 화면에서 권한설정하는 곳으로 가서 Scopes에서 권한을 부여한다.

5. 소켓모드 활성화

   Slack API 소켓모드란? 실시간 메시징 및 이벤트 기능을 구현하는 데 사용되는 기술로, 기본적으로 Slack API의 웹 소켓 연결을 통해 실기산으로 메시지 및 이벤트를 주고 받을 수 있게 해주며 이를 통해 애플리케이션은 더 빠르게 사용자와 상호작용하고, 이벤트를 감지하며, 실시간 업데이트를 제공 소켓 모드를 사용하면 회사 방화벽 뒤에서 허용되지 않을 수 있는 공용 HTTP 끝점을 노출하지 않고 봇이 작업 공간에서 상호 작용이 가능

6. 워크스페이스에 앱 설치 처음 만들고 나면 Install to Workspace에 눌러 자신이 원하는 워크스페이스에 등록하면 된다. 권한 수정과 같이 수정하였을 때는 Reinstall을 이용하여 수정 해주면 된다.

7. 채널에 봇 추가하기

@봇이름 을 하여 채널에 참여 시킨다.

Airflow 세팅

airflow를 켠 후 admin - Variavles 설정

key값에 본인이 쓸 token 변수명을 적고, val에 토큰값을 입력 (xoxb로 시작하는 토큰값)

slack_nofitications.py 작성

여러 개의 dag에 쉽게 적용 시키기 위해서 slack_nofitications.py 작성

from slack_sdk import WebClient
from datetime import datetime

class SlackAlert:
    # 클래스 인스턴스 초기화
    # 채널정보와 slack 인증 토큰을 인자로 받음
    def __init__(self, channel, token):
        self.channel = channel
        self.client = WebClient(token=token)

    def success_msg(self, msg):
        # 성공메시지를 작성하고 일자와 task id, dag id, log url을 슬랙 메세지로 출력
        text = f"""
            date : {datetime.today().strftime('%Y-%m-%d')}
            alert : 
                Success! 
                    task id : {msg.get('task_instance').task_id}, 
                    dag id : {msg.get('task_instance').dag_id}, 
                    log url : {msg.get('task_instance').log_url}
            """
        self.client.chat_postMessage(channel=self.channel, text=text)

    def fail_msg(self, msg):
        # 실패메시지를 작성하고 일자와 task id, dag id, log url을 슬랙 메세지로 출력
        text = f"""
            date : {datetime.today().strftime('%Y-%m-%d')}  
            alert : 
                Fail! 
                    task id : {msg.get('task_instance').task_id}, 
                    dag id : {msg.get('task_instance').dag_id}, 
                    log url : {msg.get('task_instance').log_url}
        """
        self.client.chat_postMessage(channel=self.channel, text=text)

내 dag에 적용하기

앞서 slack_notifications.py를 이용하여 SlackAlert 함수 사용 토큰 값에 노출을 막기 위해 Variable 지정

Variable.get을 통해 지정해두었던 slack_api_token을 불러옴 SlackAlert에 첫째 인자는 메시지를 보낼 채널명을 입력하고, 두번째 인자에는 slack token을 입력

dag에 on_success_callback과 on_failure_callback에 각각 slack_notifications.py에서 만든 함수인 sucess_msg, fail_msg를 넣어준다.

성공화면

message 채널에 notice-bot이 성공 여부를 알리는 메시지를 보내준다.

Raw data를 분석하기 위해 데이터 파싱 및 전처리를 한 후 파케이 파일로 logs_weapon에 넣는 것이 목적 logs_weapon파일에 가보니 잘 들어온 것을 확인할 수 있었음 gcp 스토리지에도 파케이 파일로 잘 들어온 모습

파케이 파일을 꺼내서 본 모습

참고 blog https://www.twilio.com/blog/how-to-build-a-slackbot-in-socket-mode-with-python

Spark 설치 후 SparkSession 띄우기

이전에 작성했던 걸 보고 client에서 PySpark를 설치했다. 설치 후 datanode1~5에 spark 파일을 복사하였다. Spark 설치하기 <-

spark_encore.tar.gz로 압축하여 각 노드에 ssh로 접속 후 압축 해제하는 과정을 진행했다. scp ./spark_encore.tar.gz datanode1:/home/hadoop/ scp ./spark_encore.tar.gz datanode2:/home/hadoop/ scp ./spark_encore.tar.gz datanode3:/home/hadoop/ scp ./spark_encore.tar.gz datanode4:/home/hadoop/ scp ./spark_encore.tar.gz datanode5:/home/hadoop/

SPARK_HOME과 Jupyter를 웹브라우저에서 접속할 수 있도록 인터페이스 설정

vim .bashrc 수정

export SPARK_HOME="/home/hadoop/spark" export PYSPARK_DRIVER_PYTHON=jupyter export PYSPARK_DRIVER_PYTHON_OPTS='notebook --ip=0.0.0.0'

주피터 노트북으로 스파크 세션이 뜨지 않는 오류

client:8888에 접속하여 jupyter notebook을 켜서 from pyspark.sql import SparkSession 명령어를 실행하면 실행이 되지 않는 현상 발생

• 현상 이유 : excutor 5개가 뜨는 시간이 너무 오래 걸려 timeout 에러 발생

기존에 스파크 드라이버 메모리를 4096 MB로 설정했었는데, 8GB로 늘리고, EXCUTOR_MEMORY도 2048 MB에서 4GB로 늘렸다.

pyspark --master yarn --num-executors 1로 줄여서 실행하여 보니 실행되었다.

설정을 그대로 하여

pyspark --master yarn --num-executors 5도 실행해봤다.

나중에 알아본 결과 이건, python3으로 단일노드로 실행한 것이었다. 내가 원하는 것은 서버를 구성한 클러스터를 모두 사용하여 분산 처리를 하려고 했던 것이었다. pyspark 커널을 띄워 pyspark를 제대로 사용하는 건 다음 포스트에 작성하려고 한다.

추후에 할 일

• 현재 프로젝트 진행 중인 데이터를 hdfs에 담아서 spark로 병렬처리해보기
• 온프레미스 환경으로 구축된 클러스터를 작업 중인 컴퓨터와 같은 와이파이를 사용하지 않아도 접속가능하도록 만들기 (open VPN)

개요

윈도우 8대 노트북을 우분투 22.04 LTS를 이용하여 각각의 서버로 만들어 하둡 구성하기 최종 프로젝트를 준비하던 중 데이터가 많아 결과를 보기까지가 오래걸려 분산처리를 한다면 얼마나 시간을 단축할 수 있을까?에 대한 생각으로 온프레미스 환경으로 노트북 8대를 연결하여 Hadoop 클러스터를 구성하였다. 데이터를 HDFS에 밀어넣어 PySpark를 통해 데이터 전처리와 분석을 해보려 한다.

🛠️ 노트북 스펙 OS : Windows 10 HOME 프로세서 : Intel(R) Core(TM) i5-7200U CPU RAM : 16GB

🛠️ 설정 환경 OS : Ubuntu 22.04 Hadoop : 3.2.1 jdk : 1.8.0 spark : 3.2.4 python : 3.10.10 (miniconda 환경)

프로젝트 인프로 설계도 (변경 가능)

목표: 여러 대의 노트북으로 하둡 클러스터 구성해보기 8개의 노트북으로 client, namenode, secondnode, datanode 1~5를 구성

Ubuntu 22.04 설치

노트북에 기존에 깔려있던 윈도우 운영체제 삭제 후, Ubuntu 운영체제 설치 (USB 이용)

※ 컴퓨터 몇 대가 USB를 통해 Ubuntu 운영체제를 설치하려고 해도 설치되지 않는 현상 발생 기존에 windows 운영체제 파티션 충돌 문제로 인해 설치가 되지 않았다. 수동으로 파티션에 있는 걸 모두 비어준 다음 설치하니 제대로 작동이 되었다.

우분투 한글 설정 방법

터미널을 열어 ibus-setup 명령어 입력

1. Settings > Keyboard > + 버튼을 누르고 > Korean 검색 후 > Korean(Hangul)을 Add

2. 여기까지 진행하면 우측상단에서 Korean(Hangul)이 추가되어서 수정할 수 있다.

   • 한글, 영어 수정 방법은 Shift+Space 키 혹은 한/영 키를 누르기

hadoop user 생성하기

sudo adduser hadoop hadoop 계정 생성

비밀번호는 hadoop으로 설정

SSH 접속을 위한 키 생성

노트북끼리 서로 원활하게 소통하기 위해 키 생성 퍼블릭 키를 통해 서로의 노트북에 접속 가능케 하기 위함

ssh-keygen -t rsa 키 생성

SSH이란? (What is SSH?) 시큐어 셀(Secure SHell, SSH)은 네트워크 상의 다른 컴퓨터에 로그인하거나 원격 시스템에서 명령을 실행하고 다른 시스템으로 파일을 복사할 수 있도록 해주는 응용 프로그램 또는 그 프로토콜을 가리킨다.

서로 ssh 명령어로 접속이 가능하도록 openssh-server 설치 sudo apt install openssh-server

호스트 이름 변경

노트북 8대를 client, namenode, secondnode, datanode1, datanode2, datanode3, datanode4, datanode5 설정

sudo hostnamectl set-hostname client sudo hostnamectl set-hostname namenode sudo hostnamectl set-hostname secondnode sudo hostnamectl set-hostname datanode1 sudo hostnamectl set-hostname datanode2 sudo hostnamectl set-hostname datanode3 sudo hostnamectl set-hostname datanode4 sudo hostnamectl set-hostname datanode5

root계정으로 hadoop 계정 권한 설정

sudo visudo 명령은 리눅스 시스템에서 /etc/sudoers 파일을 수정하기 위해 사용되는 명령

sudo visudo

root 밑에 hadoop 계정 추가해주기 계정을 추가하면 sudo 명령어를 사용할 수 있게 됨

방화벽 해제

방화벽을 해제하여 서로 연결하는데 방해요소가 없도록 설정

sudo apt install ufw 방화벽 설치

sudo systemctl stop ufw 방화벽 멈추기

sudo systemctl status ufw 방화벽 상태 확인

Ip 고정 설정

노트북에 전부 랜선을 꽂을 환경이 되지 않아 Wi-Fi로 접속하였고, 서로 지속적으로 통신을 해야 되기 때문에 IP주소를 고정하였다.

노드 간 통신 설정

퍼블릭 키를 authorized_keys에 적어 scp 명령어를 통해 모든 노드에 복사 권한 설정을 하여 서로 ssh를 통해 접속 가능하게 설정

cd ~/.ssh .ssh로 이동

cat id_rsa.pub >> authorized_keys authorized_keys에 노트북 7대에 퍼블릭 키를 받아 작성

chmod 600 ./authorized_keys 소유자에게만 읽기,쓰기 권한 부여

scp ./authorized_keys client:/home/hadoop/.ssh/ scp ./authorized_keys namenode:/home/hadoop/.ssh/ scp ./authorized_keys secondnode:/home/hadoop/.ssh/ scp ./authorized_keys datanode1:/home/hadoop/.ssh/ scp ./authorized_keys datanode2:/home/hadoop/.ssh/ scp ./authorized_keys datanode3:/home/hadoop/.ssh/ scp ./authorized_keys datanode4:/home/hadoop/.ssh/ scp ./authorized_keys datanode5:/home/hadoop/.ssh/

7대에 퍼블릭 키를 모아둔 파일을 모든 노드에 복사

위의 명령어를 이용하여 모든 노드에서 키 공유

sudo apt install vim vim 설치

vim /etc/hosts host 파일 수정

192.168.80.14 datanode1 192.168.80.150 datanode2 192.168.80.169 datanode3 192.168.80.160 datanode4 192.168.80.155 datanode5 192.168.80.170 client 192.168.80.4 namenode 192.168.80.28 secondnode

↑ 추가

ssh를 통해 모든 컴퓨터에 접속 가능

ssh client ssh namenode ssh secondnode ssh datanode1 ssh datanode2 ssh datanode3 ssh datanode4 ssh datanode5

JAVA 설치 (open jdk 1.8.0)

하둡이 자바로 구성돼있기 때문에 java 설치

sudo apt-get update

sudo apt-get install openjdk-8-jdk

java -version 버전 확인

JAVA_HOME 설정

readlink -f $(which java) JAVA가 설치된 경로 확인

sudo vi /etc/profile profile 편집 // .bashrc에 설정해도 된다.

export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64 자바 홈 경로 추가 export PATH=$PATH:$JAVA_HOME/bin 경로 지정

• :$JAVA_HOME/bin 의미 : 기존 PATH에 JAVA_HOME을 추가
  • $PATH를 제외 하고 입력할 시에 PATH가 JAVA_HOME만 남기 때문에 JAVA는 이상이 없지만 다른 명령어는 먹히지 않음

.bashrc와 .profile의 차이점

• .bashrc : 터미널에서 실행될 때마다 실행되는 파일로서, 새로운 쉘을 시작할 때마다 사용자의 환경을 설정. 사용자의 특정 프로그램에 관한 환경 변수를 정의하거나 에일리어스(Alias; 단축 명령어)를 설정하는 등의 작업을 수행. 이 파일은 로그인이 아닌 쉘에서 실행됩니다.
• .profile : 사용자가 로그인할 때 실행되는 파일. 로그인 쉘에서만 실행되며, 사용자의 환경을 설정하는 중요한 파일. 이 파일에서는 PATH 환경변수를 정의하거나, PS1환경 변수를 설정하고, 로그인 후 실행해야 할 명령어를 작성할 수 있음

일반적으로 .profile 파일은 종료되면 재시작해야하는 작업을 수행하는 것과 같은 시스템 전체의 기본적인 설정을 정의하고, .bashrc 파일은 로그인 후 사용자가 사용 가능한 작업을 정의하는 데 사용됩니다.

miniconda 설치

wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh 미니콘다 설치(wget)

sh ./Miniconda3-latest-Linux-x86_64.sh 쉘스크립트 설치 파일 실행

source ~/.bashrc 적용

Hadoop 설치

hadoop 계정 cd ~에서

wget https://archive.apache.org/dist/hadoop/common/hadoop-3.2.1/hadoop-3.2.1.tar.gz 하둡 파일 다운로드

tar xzf hadoop-3.2.1.tar.gz hadoop tar 파일 압축 해제

mv ./hadoop-3.2.1 ./hadoop 파일명 hadoop으로 변경

하둡 설정

hadoop의 conf 디렉토리에 있는 파일들을 수정하였다. 아래 파일들을 설정했다.

cd ~에서 다른 노트북에도 hadoop 설치를 위해 파일 scp를 이용해 복사

scp -r ./hadoop namenode:/home/hadoop/

• 쉘 스크립트 만들어보기

• 여러 노드에 명령어를 계속치지 않으려 쉘 스크립트로 만들어 보았다. (feat. chatGPT)

  #!/bin/bash
  

local_directory="./hadoop" # 로컬 디렉토리 경로 remote_hosts=("datanode2" "datanode3" "datanode4" "datanode5" "client" "namenode" "secondnode") # 원격 호스트 이름 > 목록 remote_directory="/home/hadoop/" # 원격 디렉토리 경로

for host in "${remote_hosts[@]}"; do scp -r "$local_directory" "$host:$remote_directory" done

```

한개의 파일에 경우는 scp로 만으로도 복사가 되지만, 대량의 파일을 전송할 경우에는 -r 옵션을 사용해야 함

추후 진행할 작업

1. pyspark 설치
2. jupyter 설치
3. HDFS에 데이터 업로드

from pyspark.sql import SparkSession

from pyspark.sql.types import StructField, StructType, StringType, LongType

스키마는 DataFrame의 컬럼명과 데이터 타입을 정의

• CSV나 JSON 같은 일반 텍스트 파일을 사용하면 다소 느릴 수 있음
• 하지만 Long 데이터 타입을 Integer 데이터 타입으로 잘못 인식하는 등 정밀도 문제가 발생할 수 있음
• 따라서 운영 환경에서 추출, 변환, 적재를 수행하는 ETL 작업에 스파크를 사용한다면 직접 스키마를 정의해야 함
• ETL 작업 중에 데이터 타입을 알기 힘든 CSV나 JSON 등의 데이터소스를 사용하는 경우 스키마 추론 과정에서 읽어 들인 샘플 데이터의 타입에 따라 스키마를 결정해 버릴 수 있음

스키마는 여러 개의 StructField 타입 필드로 구성된 StructType 객체

• StructField는 이름, 데이터 타입, 컬럼이 값이 없거나 null 일 수 있는지 지정하는 불리언값을 가짐
• 필요한 경우 컬럼과 관련된 메타데이터를 지정할 수 있음

• pyspark.sql.types: PySpark에서 구조화된 데이터를 다루기 위한 데이터 타입 및 스키마를 정의하는 모듈

• StructField: 구조화된 데이터의 필드 또는 열(column)을 정의하는 클래스입니다. 각 필드는 이름(name), 데이터 타입(dataType), 널 허용 여부(nullable) 등의 속성을 가지고 있다.
• StructType: 구조화된 데이터의 스키마(schema)를 정의하는 클래스입니다. 스키마는 여러 StructField 객체로 구성되어 있으며, 각 필드의 이름과 데이터 타입을 정의하여 DataFrame의 열 구조를 설명
• StringType: 문자열 데이터 타입을 나타내는 클래스입니다. PySpark에서 문자열은 StringType으로 표현
• LongType: 정수형 데이터 타입 중 하나인 Long을 나타내는 클래스입니다. PySpark에서 64비트 정수는 LongType으로 표현

HDFS에 csv파일에 밀어넣기

filezila를 통해 강사님이 주신 json폴더에 csv파일 모두 다운로드 받은 후 홈에다가 옮기기

홈에 옮긴 후 cd json

hdfs dfs -put *.csv ./json csv파일 밀어넣기

hdfs dfs -ls ./json 잘 들어갔는지 확인

Spark 실습해보기

myManualSchema = StructType([
  StructField("DEST_COUNTRY_NAME", StringType(), True),
  StructField("ORIGIN_COUNTRY_NAME", StringType(), True),
  StructField("count", LongType(), False, metadata={"hello":"world"})
])
df = spark.read.format("json").schema(myManualSchema)\
  .load("/json/2015-summary.json")

StructField 클래스는 PySpark에서 구조화된 데이터의 필드 또는 열(column)을 정의하는 데 사용됩니다. 각 필드는 이름(name), 데이터 타입(dataType), 널 허용 여부(nullable) 등의 속성을 가지고 있다.

• name: 필드의 이름을 나타내는 문자열. 필드의 이름은 해당 필드를 식별하는 데 사용
• dataType: 필드의 데이터 타입을 나타내는 클래스. 데이터 타입은 PySpark의 pyspark.sql.types 모듈에서 제공하는 클래스를 사용하여 지정할 수 있습니다. 예를 들어, 문자열 데이터 타입은 StringType()으로, 정수형 데이터 타입은 LongType()으로 지정할 수 있다.
• nullable: 필드의 널 허용 여부를 나타내는 부울 값입니다. 기본값은 True이며, True로 설정되면 해당 필드는 널(null) 값을 가질 수 있다.
• metadata: 필드의 메타데이터를 나타내는 맵입니다. 메타데이터는 추가 정보를 필드에 연결하기 위해 사용될 수 있습니다. 예를 들어, 필드에 대한 설명이나 태그 등의 정보를 메타데이터로 설정할 수 있습니다. 메타데이터는 선택적이며, 필요에 따라 사용할 수 있다.

  # 기본폼
  StructField(name, dataType, nullable=True, metadata=None)

metadata={"hello":"world"}의 해석 메타데이터는 일반적으로 키-값 형태의 맵(map)으로 표현 여기서는 hello가 키, world가 값이다. 키와 값은 문자열 형태로 지정함 메타 데이터의 키와 값은 개발자가 데이터에 대한 부가적인 정보를 제공하기 위해 사용하는 것이므로, 어떤 값을 사용할 지는 개발자의 재량에 따라 결정됨 예제에서는 설명보단, 임의의 값을 넣었음

df.explain() 명령어를 입력해 데이터프레임의 실행 계획을 출력 (아래는 출력값)

== Physical Plan == FileScan json [DEST_COUNTRY_NAME#0,ORIGIN_COUNTRY_NAME#1,count#2L] Batched: false, DataFilters: [], Format: JSON, Location: InMemoryFileIndex(1 paths)[hdfs://namenode:8020/json/2015-summary.json], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<DEST_COUNTRY_NAME:string,ORIGIN_COUNTRY_NAME:string,count:bigint>

df.show(5) df 5건만 확인

Select와 SelectExpr

select와 selectExpr

• select와 selectExpr 메서드를 사용하면 데이터 테이블에 SQL을 실행하는 것처럼 DataFrame에서 SQL을 사용할 수 있음

• sql과 같은 형태의 쿼리로 여러 칼럼을 선택해 출력 가능 df.select("dest_country_name", "origin_country_name").show(3)

from pyspark.sql.functions import expr, col, column

# 각각 expr, col, column 함수를 사용하여 dest_country_name열을 선택 후 출력
df.select( expr("dest_country_name"), col("dest_country_name"), column("dest_country_name")).show(3)

칼럼명 변경 (as, .alias)

# dest_country_name을 destination으로 변경
df.select(expr("dest_country_name as destination")).show(2)

# destination로 변경된 칼럼명을 alias를 이용해 dest_country_name로 변경
df.select(expr("dest_country_name as destination").alias("dest_country_name")).show(2)

selectExpr

selectExpr 메서드는 스파크의 진정한 능력을 보여줌

• 새로운 DataFrame을 생성하는 복잡한 표현식을 간단하게 만드는 도구
• 모든 유효한 비집계형(non-aggregating) SQL 구문을 지정할 수 있음
• 단, 컬럼을 식별할 수 있어야 함

df.selectExpr("*", "(dest_country_name = origin_country_name) as withinCountry").show()

"df" DataFrame의 모든 열을 선택하고, "dest_country_name"과 "origin_country_name"이 동일한지 비교하여 "withinCountry" 열을 추가한 뒤, 결과를 출력하는 작업을 수행합니다. "withinCountry" 열은 동일한 국가 내에서 여행한 여부를 나타내는 불리언(Boolean) 값으로 출력

df.selectExpr("avg(count)", "count(distinct(dest_country_name))").show(2)

"df" DataFrame에서 "count" 열의 평균과 "dest_country_name" 열의 고유한 값 개수를 계산한 뒤, 결과를 출력하는 작업을 수행

Spark 데이터 타입으로 변환하기

• 새로운 컬럼이 아닌 명시적인 값을 스파크에 전달해야 함

• 명시적인 값은 상수 값일 수 있고, 추후 비교에 사용할 무언가가 될 수도 있음
• 이때 literal을 사용
• 리터럴은 프로그래밍 언어의 리터럴값을 스파크가 이해할 수 있는 값으로 변환

df.withColumn("numberOne", lit(1)).show(2)

"df" DataFrame에 "numberOne"이라는 새로운 열을 추가하고, 해당 열의 모든 행에 값 1을 할당한 뒤, 결과를 출력하는 작업을 수행 withColumn 메서드 - DataFrame에 신규 컬럼을 추가하는 방법 .lit 함수를 사용하여 1이라는 상수 할당하고, 출력

VirtualBox를 켜서 가상머신 4개 창을 계속 띄우니 화면이 복잡함 bat파일을 통해 창을 백그라운드에서 실행

bat 파일이란?

• BAT 파일은 Batch 파일의 줄임말로, 윈도우 기반 컴퓨터에서 실행되는 스크립트 파일
• 이 파일 형식은 .bat 확장자를 가지며, 한 개 이상의 명령어를 포함하여 일련의 작업을 자동화할 수 있다.

메모장에 다음과 같이 입력 후 hadoop.bat로 저장

• 가상머신 이름 (hadoop client / hadoop namenode / hadoop secondnode / hadoop datanode3)

"C:\Program Files\Oracle\VirtualBox\VBoxManage.exe" startvm "hadoop client" --type headless
"C:\Program Files\Oracle\VirtualBox\VBoxManage.exe" startvm "hadoop namenode" --type headless
"C:\Program Files\Oracle\VirtualBox\VBoxManage.exe" startvm "hadoop secondnode" --type headless
"C:\Program Files\Oracle\VirtualBox\VBoxManage.exe" startvm "hadoop datanode3" --type headless

Superputty 접속 (client)

192.168.56.11 client 접속

hadoop 계정에서 별칭으로 지정한 명령어 실행

start_dfs start_yarn start_mr

start-dfs.sh - 분산 파일 시스템(Distributed File System)을 시작 start-yarn.sh - YARN(Yet Another Resource Negotiator) 시작 start-mr.sh - MapReduce 작업 이력 서버 시작

*용어정리 *

YARN YARN은 Hadoop 클러스터에서 리소스 관리와 작업 스케줄링을 담당 HDFS HDFS는 대용량의 데이터를 분산 저장하는 분산 파일 시스템 MapReduce MapReduce는 데이터 처리 작업을 분산하여 실행하는 분산 프로그래밍 모델

Spark 실행

pyspark --master yarn --num-executors 3 명령어 실행

client:8888 jupyter notebook 접속

# 스파크 세션 띄우기
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("PySpark to MySQL").config("spark.jars", "mysql-connector-java-8.0.21.jar").getOrCreate()

코드 해석

• SparkSession.builder : SparkSession을 생성하기 위한 빌더 객체를 생성
• appName("PySpark to MySQL") : Spark 애플리케이션의 이름을 "PySpark to MySQL"로 지정
• config("spark.jars", "mysql-connector-java-8.0.21.jar") : Spark의 환경 설정 중 "spark.jars" 옵션을 설정하여, "mysql-connector-java-8.0.21.jar"라는 JAR 파일을 Spark에 제공
  • JAR 파일은 MySQL과의 연결을 위해 필요한 JDBC 드라이버
• getOrCreate() : 설정된 옵션을 기반으로 SparkSession 객체를 생성하거나, 이미 존재하는 SparkSession 객체를 반환함. 이미 생성된 SparkSession이 있는 경우 새로운 SparkSession을 생성하지 않고 기존의 SparkSession을 사용

df = spark.read.format("csv")\
  .option("header", "true")\
  .option("inferSchema", "true")\
  .load("/data/retail-data/*")

코드 해석

• spark.read.format("csv") : SparkSession을 사용하여 CSV 파일을 읽기 위한 데이터소스 형식을 설정
• .option("header", "true") : CSV 파일의 첫 번째 줄을 헤더로 처리하도록 옵션을 설정
• .option("inferSchema", "true") : 데이터프레임의 스키마를 자동으로 추론하도록 옵션을 설정
• .load("/data/retail-data/*") : 지정된 경로에서 CSV 파일을 로드하여 데이터프레임을 생성 /data/retail-data/*는 /data/retail-data/ 디렉토리에 있는 모든 CSV 파일을 로드하도록 지정

# PySpark에서 데이터프레임을 임시 뷰로 등록하는 코드
df.createOrReplaceTempView("dfTable")

# dfTable에 있는 모든 칼럼을 보여줘 (상위 5개만 표시)
spark.sql("select * from dfTable").show(5)

데이터 베이스와 자료 교환해보기

JDBC(Java Database Connectivity)란?

• JDBC는 Java 프로그래밍 언어를 사용하여 데이터베이스와 연결하고 상호 작용하기 위한 API(응용 프로그래밍 인터페이스)입니다. JDBC를 사용하면 Java 애플리케이션에서 다양한 데이터베이스 시스템에 접속하여 데이터베이스와 통신할 수 있다.
• JDBC를 사용하면 데이터베이스 관리 시스템(DBMS)에 대한 특정 드라이버를 로드하고, 연결을 설정하고, SQL 쿼리를 실행하고, 결과를 검색하는 등의 작업을 수행할 수 있다. JDBC는 일관된 방식으로 다양한 데이터베이스와 상호 작용할 수 있도록 표준화된 API를 제공한다.

# 우리 조의 데이터 베이스의 접속 (테이블명은 battleGround)
jdbc_url = "jdbc:mysql:// [ip주소]:3306/[테이블명]?serverTimezone=UTC"  # 데이터베이스 URL
table_name = "jk"  # 테이블 이름
properties = {
    "user": "[사용자명]",  # MySQL 사용자명
    "password": "[비밀번호]"  # MySQL 비밀번호
}

코드해석

• jdbc_url: 데이터베이스의 URL을 나타냅니다. 주어진 URL은 jdbc:mysql://[IP주소]:3306/battleGround?serverTimezone=UTC로, MySQL 데이터베이스에 접속하기 위한 URL입니다. [IP 주소]는 데이터베이스 서버의 IP 주소이고, 3306은 MySQL의 기본 포트 번호입니다. battleGround은 데이터베이스의 이름을 나타냅니다. serverTimezone=UTC는 데이터베이스 서버의 시간대를 UTC로 설정
• table_name: 작업을 수행할 테이블의 이름을 나타냅니다. 주어진 코드에서는 jk라는 테이블을 사용합니다.
• properties: 데이터베이스 연결을 위한 추가 속성을 포함하는 맵입니다. 여기서는 user와 password를 설정하고 있습니다. user는 MySQL의 사용자명을 나타내며, password는 해당 사용자의 비밀번호를 나타냅니다.
• 이러한 설정을 사용하여 코드에서는 주어진 JDBC URL을 통해 MySQL 데이터베이스에 연결하고, jk라는 테이블에서 작업을 수행할 수 있게 된다.

DataFrame을 DB에 저장하기

# 앞서 불러왔던 retail-data에 모든 csv를 battleGround DB에 만든 jk테이블로 저장
df.write.jdbc(url=jdbc_url, table=table_name, mode="append", properties=properties)

명령어 이후 DB 확인 (아래 사진)

DB에 있는 데이터 불러오기

battleGround DB에 있는 position 테이블 읽기
battle_position = spark.read.jdbc(url=jdbc_url, table='position', properties=properties)

Vagrant란?

가상화(Virtualization)는 실제 운영체제 위에 가상화 소프트웨어를 설치한 후에 소프트웨어를 통해 하드웨어(CPU, Memory, Disk, NIC 등)를 에뮬레이션한 후에 이 위에 운영체제(Guest OS)를 설치하는 것을 의미합니다. 가상화를 해 주는 소프트웨어를 하이퍼바이저(Hypervisor) 라고 하며 종류로는 이 책에서 사용하는 VirtualBox, 그리고 VMWare, Xen 등이 있습니다.

• 에뮬레이션은 컴퓨터 또는 기타 여러 주변 장치의 기능을 다른 컴퓨터에서 구현하는 것을 의미
• 프로그래밍 언어: 루비

Vagrant 설치

아래 사이트에 접속하여 윈도우 버전을 설치

CMD창 열기

mkdir test test파일을 만들어 vagrant init vagrant 초기화

Vagrantfile 파일을 열어서 config.vm.box = 에 내용을 깔고자 하는 centos7 입력 후 저장

이후 vagrant up 명령어를 입력하면 centos 7 가상머신이 하나 만들어짐

vagrant ssh 가상머신 접속

vagrant halt 가상 머신 종료

hostnamectl set-hostname client clinet로 이름 변경

sestatus 운영체제 보안 설정 확인

sudo sed -i 's/^SELINUX=enforcing$/SELINUX=permissive/' /etc/selinux/config

chatGPT 명령어정리 sudo: 명령어를 관리자 권한으로 실행하기 위해 sudo를 사용합니다. 사용자에게 관리자 액세스 권한이 필요한 경우 비밀번호를 입력하라는 프롬프트가 나타날 수 있습니다. sed -i 's/^SELINUX=enforcing$/SELINUX=permissive/' /etc/selinux/config: sed는 텍스트 스트림에서 패턴을 찾아 다른 패턴으로 치환하는 유틸리티입니다. 이 명령어는 /etc/selinux/config 파일에서 "SELINUX=enforcing"이라는 패턴을 찾아 "SELINUX=permissive"로 변경합니다. -i 옵션은 변경된 내용을 원본 파일에 바로 적용하도록 지시합니다.

hadoop 계정 생성 adduser hadoop 하둡 유저 생성 passwd hadoop 패스워드 설정

su hadoop 하둡 유저 진입 ssh-keygen -t rsa RSA 알고리즘을 사용하여 SSH 키를 생성

cat id_rsa.pub >> authorized_keys SSH 서버에서 사용자 인증을 위해 허용되는 공개키를 저장하는 파일

처음에 들어가보면 ~가 아니라 root로 설정이 되어있는데, ~로 오도록 설정

vim ~/.bashrc 편집기에서 cd ~ 추가 후 wq

root에서 권한 부여

• visudo

  :100 백번째 줄로 가서 101번 라인에 아래 추가 hadoop ALL=(ALL) ALL 추가 hadoop유저에 root 권한 부여

hadoop Java 설치

sudo yum install java-1.8.0-openjdk ant -y

cd /usr/lib/jvm ls

자바 홈 설정

vim ~/.bashrc

export JAVA_HOME="/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.372.b07-1.el7_9.x86_64"

wget 설치

sudo yum install wget cd ~ 에서

하둡 파일 설치

wget https://archive.apache.org/dist/hadoop/common/hadoop-3.2.1/hadoop-3.2.1.tar.gz

tar xzf hadoop-3.2.1.tar.gz 압축해제

mv ./hadoop-3.2.1 ./hadoop hadoop으로 이름바꾸기

FileZilla에서

/home/hadoop/hadoop/etc/hadoop 경로에 강사님이 올려주신 5개 파일 넣기

vim ~/.bashrc 에서 아래 내용 추가 hadoop설정 - 전에 ec2환경에서 만든 하둡 환경 복사해서 만듦

export SPARK_HOME=/home/hadoop/spark
export HADOOP_HOME=/home/hadoop/hadoop
export HIVE_HOME=/home/hadoop/hive
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
export HADOOP_INSTALL=$HADOOP_HOME
export HADOOP_MAPRED_HOME=$HADOOP_HOME
export HADOOP_COMMON_HOME=$HADOOP_HOME
export HADOOP_HDFS_HOME=$HADOOP_HOME
export HADOOP_YARN_HOME=$HADOOP_HOME
export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native
export HADOOP_OPTS="-Djava.library.path=$HADOOP_HOME/lib/native"
export PATH=$PATH:$HADOOP_HOME/sbin:$HADOOP_HOME/bin:$HIVE_HOME/bin

하둡 이미지를 굽기 위해 종료

• 현재 바로 off sudo shutdown -h now

이미지 복제 방법 1

하둡을 누른 후 복제를 이용하여 복사

이미지 복제 방법 2

OCI로 내보내기 - 설정 그대로 해서 생성 가상 시스템 가져오기 해서 설정 내보낸 거 불러오기

방법 2가지 중 하나 선택하여 4개 만들기 각각의 이름 Client, namenode, Secondnode, Datanode3으로 지정하기

실행 시 오류 메세지를 보고 설치 c++ 2019 redistributable 검색 후 파일 다운받기

CentOS 설치

Virtual Box 실행

하둡 설치를 위한 설정하기

가상 머신 만들기 이름 hadoop

iso image 다운로드 받은거 넣기

ip addr

vim 설치

네트워크 설정

vim ifcfg-enp0s8

고정ip 설정 아래 내용 추가 수정한 부분 - 노란색

명령어정리 ft. chatGPT

• "bootproto=none"을 설정하면 해당 시스템은 네트워크 부팅 프로토콜을 사용하지 않는다. 즉, 시스템은 로컬 저장 장치에서 운영체제를 부팅하여 시작하게 된다. 이 설정은 일반적으로 개별 컴퓨터에서 사용되며, 네트워크 부팅이 필요하지 않은 경우에 유용하다.
• IPADDR = 192.168.56.11: 이 설정은 해당 네트워크 인터페이스에 할당된 IP 주소를 나타냅니다. 여기서는 192.168.56.11이라는 IP 주소가 할당되었다는 것을 의미합니다. 이 IP 주소는 해당 네트워크에서 해당 인터페이스를 식별하는 데 사용됩니다.
• NETMASK = 255.255.255.0: 서브넷 마스크는 네트워크 ID와 호스트 ID를 구분하기 위해 사용되는 값입니다. 여기서는 255.255.255.0이라는 서브넷 마스크가 설정되었습니다. 이것은 해당 네트워크의 IP 주소에서 처음 24비트가 네트워크 ID를 나타내고, 나머지 8비트가 호스트 ID를 나타내는 것을 의미합니다.
• GATEWAY = 192.168.56.1: 게이트웨이는 네트워크 간에 데이터를 전송하는 데 사용되는 네트워크 장치입니다. 여기서는 192.168.56.1이라는 IP 주소가 게이트웨이로 설정되었습니다. 이것은 해당 네트워크에서 다른 네트워크로 데이터를 전송할 때 패킷이 이 IP 주소를 통해 라우팅되도록 설정되었다는 것을 의미합니다.
• "ONBOOT=YES"를 설정하면 해당 네트워크 인터페이스가 자동으로 활성화되어 네트워크 서비스에 연결됩니다. 이렇게 하면 사용자가 수동으로 인터페이스를 활성화하지 않고도 시스템이 부팅될 때마다 네트워크 연결이 자동으로 설정됩니다.

설정후에 wq 저장

ip addr을 보면 enp0s8에 설정해둔 192.168.56.11/24가 설정돼있는걸 확인할 수 있음

ping 명령어를 통해 통신되는거 확인

putty 접속

용어 정리

• IP 주소 (IP address): IP 주소는 인터넷 프로토콜 (IP)을 사용하는 컴퓨터나 네트워크 장치를 식별하는 데 사용되는 숫자로 된 주소입니다. IP 주소는 네트워크 상에서 각각의 기기에 고유하게 할당됩니다. IP 주소는 IPv4 (32비트)와 IPv6 (128비트) 두 가지 버전이 있습니다.
• 서브넷 마스크 (Subnet Mask): 서브넷 마스크는 IP 주소를 네트워크 부분과 호스트 부분으로 구분하는 데 사용되는 숫자로 된 값입니다. 서브넷 마스크는 IP 주소의 네트워크 ID 부분을 나타내는 역할을 합니다. 예를 들어, 192.168.0.0/24의 경우, 24비트 서브넷 마스크를 사용하고 있으며, 처음 24비트가 네트워크 ID를 나타내고 나머지 8비트가 호스트 ID를 나타냅니다.
• 게이트웨이 (Gateway): 게이트웨이는 네트워크 간에 데이터를 전송하는 데 사용되는 네트워크 장치입니다. 게이트웨이는 패킷을 받아서 목적지 네트워크로 전달하거나 다른 네트워크로 경로를 설정하는 역할을 합니다. 일반적으로 게이트웨이는 네트워크의 첫 번째 IP 주소로 설정되며, 로컬 네트워크 외부로 나가는 패킷을 전달하는 역할을 수행합니다.
• CIDR 표기법 (CIDR notation): CIDR 표기법은 IP 주소와 네트워크 세그먼트를 표현하는 데 사용되는 표기법입니다. 예를 들어, "192.168.0.0/24"는 IP 주소가 192.168.0.0이며, 24비트의 서브넷 마스크를 가지는 네트워크를 의미합니다. CIDR 표기법은 IP 주소와 서브넷 마스크를 조합하여 네트워크를 명확하게 식별하는 데 도움을 줍니다.

putty 접속

• 순서대로 실행 su hadoop start_dfs start_yarn start_mr

• Spark 실행 nohup pyspark --master yarn --num-executors 3 & 웹사이트에 client:8888로 접속 (client가 탄력적 ip가 아니라면 ec2에서 퍼블릭 Ipv4 주소 복사후 :8888 해주기

주피터 노트북 켠 후 아래 명령어 입력 ↓ from pyspark.sql import SparkSession

명령어 입력 후 client:4040 접속시 아래 화면 볼 수 있음

실행중인 프로세스 확인해보기 ssh [NodeName] jps namenode, secondnode, datanode3 확인해보면 아래와 같이 실행되고 있음

스파크 아키텍처(Spark Architecture)

데이터셋 준비

cd ~ 에서 wget https://mydatahive.s3.ap-northeast-2.amazonaws.com/mnm_dataset.csv 파일 다운로드

• hdfs dfs 별명 붙이기 (계속 적을 때 간단히 사용하기 위해서) vim ~/.bashrc alias hd="hdfs dfs" 약어 입력해주기 source ~/.bashrc 적용 hd -ls / 확인

hd -put mnm_dataset.csv /mydata/ 로컬 파일 시스템에서 Hadoop 분산 파일 시스템(HDFS)로 파일을 업로드하기 위해 사용되는 HDFS 명령어

hd -ls /mydata 파일 업로드 확인

Jupyter Notebook

• from pyspark.sql import SparkSession 스파크세션 실행

• mnm_df = spark \

    .read \
    .format("csv") \
    .option("header", "true") \
    .option("inferSchema", 'true') \
    .load("/mydata/mnm_dataset.csv")

  read format - csv 형식의 파일 불러오기 / option - 첫줄이 헤더, 스키마는 스파크가 알아서 설정 / load 뒤에는 경로

Spark의 특징 지연연산/액션

• mnm_df.show(n=5) 액션 명령을 내려야 결과 표시

• 이후 실습

  mnm_df에서 각 주의 색상별 합계를 계산한 다음, 합계를 기준으로 오름차순 정렬

mnm_df에서 캘리포니아(CA) 주의 색상별 합계를 계산한 다음, 합계를 기준으로 오름차순 정렬

• 다른 예제

putty client cd ~에서 wget https://mydatahive.s3.ap-northeast-2.amazonaws.com/fakefriends.csv 예제 파일 받기

하둡 디렉토리에 업로드 시키기 hd -put fakefriends.csv /mydata/

Jupyter notebook

파일 불러오기 첫줄이 헤더는 아니고, 스키마는 알아서

열이름 바꾸기 selectExpr

age별 count 평균

df_result로 변수 저장후 result.csv파일로 저장

namenode:50070 파일디렉토리에서 파일 확인

hd -cat /mydata/result.csv/* 터미널에서 확인하기

wget https://dlcdn.apache.org/spark/spark-3.2.4/spark-3.2.4-bin-hadoop3.2.tgz 해서 다운로드 받기

tar xzf spark-3.2.4-bin-hadoop3.2.tgz 압축 해제

mv ./spark-3.2.4-bin-hadoop3.2 ./spark 이름 바꾸기

conf 파일 업로드

/home/hadoop/spark/conf 경로에 강사님이 올려주신 파일 3개 넣기

• Fillzilla 이용

bash.rc 설정

• 스파크 환경변수, 실행경로 설정

설정후 source ~/.bashrc써서 적용시키기

spark 실행

pyspark 명령어를 실행

exit()로 나오고 다음 세팅 준비

jupyter 세팅

pip install jupyter  #주피터 설치
jupyter notebook --generate-config  

• jupyter notebook --generate-config
  • jupyter notebook --generate-config 명령어를 실행하면 기본값으로 설정된 Jupyter Notebook 설정 파일(jupyter_notebook_config.py)이 생성
  • 이 파일이 생성되면 사용자는 해당 파일을 편집하여 Jupyter Notebook의 IP 주소, 포트 번호, 실행 모드(인라인 플롯 등), 보안 설정(비밀번호, SSL, 토큰 등) 및 기타 설정을 변경할 수 있음

jupyter 실행

ipython 실행

from notebook.auth import passwd

• 자신의 노트북 암호 만들기 passwd() - 자신의 원하는 암호 입력 / 입력하고 나오는 문자열 복사해두기(이후에 편집기에서 넣어야 됨) vim ~/.jupyter/jupyter_notebook_config.py 편집기 아래와 같이 수정

136번 줄 : 모든 도메인에서 접속 가능하도록 설정 450번 줄 : 노트북 디렉토리 설정 458번 줄 : 주피터 노트북이 시작될 때 웹 브라우저를 자동으로 열지 않도록 함 469번 줄 : 주피터 노트북에 암호를 설정 (아까 복사해둔 문자열 복사)

작업폴더 만들기

mkdir workspace

pyspark --master yarn --num-executors 3

• pyspark --master yarn --num-executors 3 명령어는
  • pyspark: PySpark 명령어로, PySpark 애플리케이션을 실행한다.
    • PySpark은 Apache Spark를 사용하여 파이썬으로 효율적인 빅 데이터 처리를 가능하게 해준다.
  • --master yarn: 이 옵션은 애플리케이션의 클러스터 관리자를 YARN(Yet Another Resource Negotiator)으로 설정합니다.
    • YARN은 클러스터에서 자원 관리를 담당하며, 맵리듀스 및 다양한 애플리케이션 프레임워크가 하둡에서 실행될 수 있도록 한다.
  • --num-executors 3: 이 옵션은 클러스터에서 사용할 Spark Executor의 개수를 3으로 설정한다.
    • Executor는 Spark 애플리케이션에서 작업 작업(load data, process, store)을 수행하는 프로세스

nohup pyspark --master yarn --num-executors 3 &

• nohup: 터미널이 종료된 후에도 계속 실행되도록 설정, 터미널과의 연결이 끊겨도 프로세스가 중단되지 않음
• &: 주어진 명령어를 백그라운드에서 실행

위 명령어 실행 후 화면에 아래와 같은 것이 보이면 http://client:8888로 들어간다.

이후, 아래 명령어 실행

from pyspark.sql import SparkSession

SparkSession

• SparkSession은 Apache Spark 애플리케이션을 시작하는데 필요한 프로그래밍 진입점(entry point)이다.

• 이를 통해 DataFrame과 Dataset의 생성과 작업을 관리하고, cluster manager와 application name, executor memory, number of executor cores 등의 Spark 애플리케이션 설정을 구성할 수 있다.
• 또한, SparkSession은 다양한 데이터 소스와 직접 연결할 수 있는 read 및 write 메서드를 지원한다.

• secondnode:8088이나 client 퍼블릭v4 ip주소(탄력적 ip설정했다면 탄력적ip):4040를 보면 아래와 같은 화면을 볼 수 있음

jupyter Directory에 spark 예제.ipynb와 hflight.csv 파일 업로드

참고

• spark는 코드를 다 만들고 실행하면 실행 계획만!
• show나 action을 통해 실행이 일어남!

용어 정리

다음시간에 사용할 파일

Parquet이란?

• 대량의 데이터를 효율적으로 저장하고 처리하기 위한 열 지향(컬럼 기반) 파일 형식이다. Parquet 파일은 대규모 데이터 처리를 위해 설계되었으며, Hadoop 기반의 분산 데이터 처리 시스템인 Apache Hadoop, Apache Spark, Apache Hive 등에서 널리 사용된다.

Parquet은 다양한 데이터 처리 작업에 적합한 형식을 제공한다. 일반적으로 데이터 웨어하우스, 비즈니스 인텔리전스 및 대규모 분석 시나리오에서 사용되며, 특히 대규모 데이터 세트에서의 쿼리 성능을 향상시킬 수 있다. Parquet 파일은 데이터를 컬럼 기반으로 저장하므로 필요한 컬럼만 읽거나 필터링할 수 있어서 읽기 작업의 효율성을 높인다. 또한 압축 및 직렬화 알고리즘을 사용하여 저장 공간을 절약하고 데이터 전송 속도를 높일 수 있다.

Hive를 이용하여 영화 크롤링 파일 mapreduce 해보기

• hadoop 유저 접속

  su hadoop
  # 별칭으로 지정했던 명령어 실행하기
  start_dfs
  start_yarn
  start_mr

  

• 제공해준 파일 tmdb.zip 파일 받기

clinet 에서 실행

cd ~
wget https://mydatahive.s3.ap-northeast-2.amazonaws.com/tmdb.zip

• unzip 설치

  sudo yum install unzip #unzip설치
  # tmdb 파일 만들어서 tmdb.zip 파일을 tmdb파일안에 압축해제
  mkdir tmdb && unzip ./tmdb.zip -d ./tmdb
  # 하둡에 tmdb파일 만들고 csv파일 붓기
  hdfs dfs -mkdir /tmdb
  hdfs dfs -put ./*.csv /tmdb

• 잘들어갔는지 확인 (namenode:50070에서 확인)

• putty에서 확인하는 방법

  hdfs dfs -ls /tmdb

  

Hive 맵리듀스 해보기

• 받았던 csv파일을 sql 쿼리를 날려 테이블 생성

CREATE EXTERNAL TABLE IF NOT EXISTS movie(
    Movie_ID  STRING,
    Adult  STRING,
    Backdrop_Path  STRING,
    Genres  STRING,
    Homepage  STRING,
    Original_Language  STRING,
    Original_Title  STRING,
    Overview  STRING,
    Popularity  STRING,
    Poster_Path  STRING,
    Production_Companies  STRING,
    Production_Countries  STRING,
    Release_Date  STRING,
    Revenue  STRING,
    Runtime  STRING,
    Spoken_Languages  STRING,
    Status  STRING,
    Tagline  STRING,
    Title  STRING,
    Vote_Average  FLOAT,
    Vote_Count  INT)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
LOCATION '/tmdb';

• 잘 적용됐는지 확인해보기

SELECT * FROM movie limit 5;

장르별로 집계해보기

SELECT Genres , AVG(Vote_Count) FROM movie GROUP BY Genres 

• map 과 reduce 과정

• 결과창

putty 접속 후

su hadoop

#bashrc 명령어 실행
$ start dfs
$ start_yarn
$ start_mr

1. hdfs에 새로운 폴더 생성

   $ hdfs dfs -mkdir /mydata

2. hdfs에 데이터 넣기

   $ hdfs dfs -put ~/hadoop/etc/hadoop/`*.xml` /mydata

3. Hadoop Cluster에서 텍스트 파일 검색하기

   $ hadoop jar ~/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.2.1.jar grep /mydata /output2 'dfs[a-z.]+'

Hadoop MapReduce의 예제 중 하나인 grep 실행 하둡 클러스터에서 텍스트 파일을 검색

/mydata 경로에 있는 텍스트 파일을 dfs[a-z.]+라는 문법 (dfs가 들어가는 거를 wordcount)을 통해서 작업할거고, 그 아웃풋을 /output2 에 담겠다.

4. MapReduce 성공 결과 세부 내역 확인 노란색 url 경로를 들어가면 아래 그림과 같이 확인할 수 있다.

5. 결과 확인

   $ hdfs dfs -cat /output2/*

cd ~/hadoop/etc/hadoop에서 vim workers

datanode1, datanode2, datanode3 입력

workers 파일 복사(scp 사용법)

scp [파일대상][서버][서버의 파일경로]

scp ./workers namenode:/home/hadoop/hadoop/etc/hadoop/ (namenode, secondnode, datanode3 모두 같은 명령어 실행)

다른 서버끼리 미리 ssh 연결되어 있어야 scp 명령어 사용 가능 이 작업을 완료하면 worker 노드를 식별하고 클러스터의 구성을 설정, 작업 분산을 할 수 있음

하둡 서버의 resource map

ec2 보안그룹 설정

인바운드 규칙 추가 hadoop이 형성된 인스턴스의 보안그룹으로 설정하면 ping datanode3을 했을 때, 핑이 날라가는 걸 확인할 수 있음

hadoop format

ssh namenode 
hadoop namenode -format #namenode 초기화

hadoop 클러스터를 처음 설정할때 또는 데이터를 삭제하고 새로운 클러스터를 구성할 때 사용 namenode에 어떤 파일이 어디에 저장되고 정보를 저장하는 장부 역할을 수행 그 장부를 초기화해서 내용을 쓸 준비를 하는 과정

start-dfs

sssh namenode
start-dfs.sh : 분산파일시스템(HDFS)의 서비스를 시작

jps : 자바프로세스를 알려주는 명령어 jps를 입력하면 Namenode, Jps, Datanode가 출력됨

start-yarn

ssh secondnode
start-yarn.sh

yarn은 지원부서같은 개념, 한 node가 빡세면 다른곳으로 보내주는 개념 jps를 입력하면 Datanode, Nodemanager, Jps, Resourcemanager, secondaryNamenode가 출력됨

start-mr

mr-jobhistory-daemon.sh start historyserver jps입력 시 Datanode, Jps, Nodemanager가 출력됨

hadoop 상태 확인 페이지

보안그룹 인바운드 규칙 추가 50070 포트로 소스는 0.0.0.0/0으로 시정하고 규칙 추가 hadoop 상태 확인하는 페이지 열기 주소창에 namenode public IP:50070 입력

하둡 로그 확인

cd hadoop/logs에서 log를 확인할 수 있음

대문자 G - 맨끝으로가기 소문자 gg - 맨 처음으로 가기

tail을 맨 앞에 붙이면 해당 파일의 끝부분만 보여줌, 끝에 3줄만 보고 싶으면 명령어 맨 뒤에 -n 3를 붙여줌

log파일을 vim으로 열어서 error가 뜨는 부분을 확인

명령어 쉽게 사용하기

ssh namenode stop-dfs.sh, ssh secondnode start-yarn.sh같은 명령어를 쉽게 사용하기 위해 별칭 설정

vim ~/.bashrc 입력하고 아래 부분을 추가
alias start-dfs="ssh namenode start-dfs.sh"
alias start-yarn="ssh secondnode start-yarn.sh"
alias stop-dfs="ssh namenode stop-dfs.sh"
alias stop-yarn="ssh secondnode stop-yarn.sh"
alias start-mr="ssh namenode mr-jobhistory-daemon.sh start historyserver"
alias stop-mr="ssh namenode mr-jobhistory-daemon.sh stop historyserver"