[Quadcore Team/Trend] 인기 트윗 랭킹

 

이제서야 올리는 프로젝트 정리 2...

 

* 인기 트윗 랭킹 추출 과정

 

1) Cassandra to Spark

• 카산드라의 raw data 중 리트윗 된 data 추출 ( 올라오는 트윗 중 80% 이상이 리트윗 데이터)
• 현재시간 기준 30초 전까지 data 추출

if __name__ == "__main__":
        while True:
            # 현재시간 마이크로 세컨즈 까지
            current_time = int(time.time() * 1000000)  # 현재시간 마이크로 세컨즈 까지
            # redis 저장 포맷 시간 형식 ( 년/월/일/시/분) 으로 
            current_time_format = datetime.datetime.fromtimestamp(int(current_time/1000000)).strftime('%Y/%m/%d/%H/%M')
            # 카산드라로부터 data 불러오기 (30초 마다)
            lines = spark.read \
            .format("org.apache.spark.sql.cassandra") \
            .options(table="master_dataset", keyspace="bts") \
            .load().select(schema)
            .where(col('timestamp') >= current_time - SECONDS)\
            .where(col('timestamp') <= current_time)\
            .where(col('retweeted') == True).limit(100).cache()
            print(current_time_format)
            print(current_time)  # 현재시간 출력

            result = process_tweet(lines)
            if result is not False:
                #print(result)
                save_tweet(result, current_time_format)
            else:
                print('there is no data')
            time.sleep(10)

 

2) Tweet Processing

 

data 가 존재할 경우

• retweeted_status 중 id, favorite_count, quote_count, retweet_count 
• retweeted_status 전체
• 해당 글이 쓰여진 시간 timestamp 값

 

뽑아내어 새로운 dataframe 생성 --> 이 과정에서 엄청 헤맸다...(따로 정리할 예정)

 

새로운 dataframe 의 StructType

struct = StructType([StructField("tweet_id", LongType(), False),
                     StructField("tweet_content", StringType(), False),
                     StructField("favorite_count", LongType(), False),
                     StructField("quote_count", LongType(), False),
                     StructField("retweeted_count", LongType(), False),
                     StructField("timestamp", LongType(), False)])

def process_tweet(data):
    print('process 들어옴')
    if bool(data.take(1)):        
        # df 생성에 필요한 데이터 추출 (리트윗 된 트윗 id, 트위 내용, 트윗 좋아요 개수, 트윗 인용 개수, 리트윗 개수, 작성시간)
        tweet_content = data.select('retweeted_status').rdd.flatMap(lambda value: value).collect()
        tweet_id = data.select('retweeted_status').rdd.map(lambda value: json.loads(value[0])) \
            .map(lambda v: v['id']).collect()
        favorite_count = data.select('retweeted_status').rdd.map(lambda value: json.loads(value[0])) \
            .map(lambda v: v['favorite_count']).collect()
        quoted_count = data.select('retweeted_status').rdd.map(lambda value: json.loads(value[0])) \
            .map(lambda v: v['quote_count']).collect()
        retweeted_count = data.select('retweeted_status').rdd.map(lambda value: json.loads(value[0])) \
            .map(lambda v: v['retweet_count']).collect()
        timestamp = data.select('timestamp').rdd.flatMap(lambda value: value).collect()

        # 새로운 dataframe 생성
        newDF = spark.createDataFrame(
            zip(tweet_id, tweet_content, favorite_count, quoted_count, retweeted_count, timestamp), struct)
        newDF.show()
        result = rank_tweet(newDF)
        return result
    else:
        return False

 

3) tweet Ranking

• 새롭게 만든 dataframe 에서 favorite, quote, retweet count 값 합산한 새로운 칼럼 total 생성
• 같은 tweet_id 중 최신 것만 집계하여 total 순으로 정렬
• 상위 10개 tweet 내용만 리스트로 변환

# 트윗 랭킹 집계
def rank_tweet(df):
    if bool(df.take(1)):
        print('rank 들어옴')
        # (favorite + quote + retweet) count 값 합산한 새로운 column 'total' 생성
        df = df.withColumn('total', df.favorite_count + df.quote_count + df.retweeted_count)
        # 같은 tweet_id 중 최신 것만 집계 (total 순으로)
        rank = df.groupBy(df.tweet_id).agg(
            f.first('tweet_content').alias('tweet_content'),
            f.first('total').alias('total'),
            f.max('timestamp').alias('timestamp')
        ).orderBy('total', ascending=False)
        rank.show()
        # 상위 랭킹 10개의 tweet 내용 리스트로 변환
        tweet_list = rank.select('tweet_content').rdd.flatMap(lambda x: x).take(10)
        return tweet_list
    else:
        print('데이터프레임 비어있음')
        return False

 

4) Save Tweet

• python list 형태의 data 를 redis 에 저장하기 위해 bytes 로 변환시킴
• 현재시간 ( 년/월/일/시/분 ) 기준: key 으로 redis에 일주일 만료 타임 지정해서 저장

 

 

결과

 

count 수가 많은 순대로 정렬한 결과