[Quadcore Team/Trend]해시태그 분석

 

미루고 미루다 이제서야 올리는 프로젝트 정리 1...

* Hashtag Ranking 과정

1) raw tweet 전처리

• raw tweet df 중 'entities.hashtags.text' 추출 ( 해시태그 column 만 추출 )
• rdd 로 만든 후 collect()를 통해 모든 row 하나의 리스트 안에 저장되도록 하기

def process(rdd):
    try:
        rawTweet = spark.read.json(rdd)
        #creates an in-memory table that is scoped to the cluster in which it was created.
        rawTweet.registerTempTable("tweets") 
        hashtag = rawTweet.selectExpr('entities.hashtags.text as hashtag').rdd.flatMap(lambda x : x)
        print(hashtag.collect())
        # 현재 타임에 들어온 hashtag 전처리
        result = hashtag_processing(hashtag.collect())

        #word count 작업을 위해 결과 rdd로 만들어줌
        rdd = spark.sparkContext.parallelize(result)
        word_count(rdd)

    except:
        pass

 

 

2) 해당 시간 안에 들어온 hashtag 모두 모아 전처리

• 각 tweet 별로 유사 단어 제거 

( ex) 멤버 이름 : 영어 - 한글 하나의 단어로 만든 후 set 을 통해 중복 제거하고 하나만 남긴다.)

 

이렇게 하는 이유 ?

- 하나의 트윗 안에서 중복 멤버 이름을 다 집계하는 것은 트렌드 취지에 맞지 않는다고 판단!

- 트렌드(다양한 유저들이 언급 많이 하는) 순위 집계에 한번만 반영하면 된다고 생각했다. → 1 트윗 1 집계

• 전체 트윗 해시태그 안에서 불용어 제거

# hashtag 전처리
    def hashtag_processing(text):
        del_similar = []
        result = []
        # 각 tweet 별 유사어 제거
        for v in text:
            if not v:
                continue
            words = '-'.join(v)
            words = words.upper()
            temp = []
            for i in similarwords[0]:
                if i in words:
                    words = words.replace(i, 'JUNGKOOK')
            for i in similarwords[1]:
                if i in words:
                    words = words.replace(i, 'JIMIN')
            for i in similarwords[2]:
                if i in words:
                    words = words.replace(i, 'JHOPE')
            for i in similarwords[3]:
                if i in words:
                    words = words.replace(i, 'SUGA')
            for i in similarwords[4]:
                if i in words:
                    words = words.replace(i, 'V')
            for i in similarwords[5]:
                if i in words:
                    words = words.replace(i, 'RM')
            for i in similarwords[6]:
                if i in words:
                    words = words.replace(i, 'JIN')
            temp = words.split('-')
            temp = list(set(temp))
            del_similar.append(temp)
    
        total = list(chain.from_iterable(del_similar))  # 리스트 안에 리스트 하나의 리스트로 합치기
        # 불용어 제거
        for i in total:
            if i not in mystopwords:
                result.append(i)
    
        return result

 

 

3) 전처리 후 집계 된 hashtag word count

• 해시태그 전처리까지 완료한 후 word count 작업을 위해 다시 rdd 로 만들어 준다.

• map, reduceByKey 활용하여 transformation 수행

  ( 이 때, 최소 3번 이상 등장한 hashtag 만 filter 적용)

   

• takeOrdered 를 통해 action 수행

  ( 이 때, 상위 20 개만 가져오도록 )

 

 

4) 최종 결과 redis에 저장 ( 클라이언트에서 요청 시 빠르게 전달하기 위해 redis 에 일주일 만료로 저장)

# 추출된 단어 word count
def word_count(list):
    print('word count 들어옴')
    pairs = list.map(lambda word: (word, 1))
    # 상위 10개만 가져오기 + 등장빈도 2번 이상
    wordCounts = pairs.reduceByKey(lambda x, y: x + y).filter(lambda args: args[1] > 2)
    ranking = wordCounts.takeOrdered(15, lambda args: -args[1])
    print(ranking)
    return ranking

    # 해시태그 순위 저장


def save_hashtag(data, time):
    # key : 'hashtag' , value : 순위 결과 json 으로 redis 저장
    rank_to_json = json.dumps(data)
    myRedis.set('hashtag', rank_to_json, ex=60 * 60 * 24 * 7)
    print('저장완료')

 

 

결과