diff --git a/ds_new_user_analy.py b/ds_new_user_analy.py index d8c0b6b..cf84da3 100644 --- a/ds_new_user_analy.py +++ b/ds_new_user_analy.py @@ -58,10 +58,10 @@ OPENSEARCH_CONFIG = { # 한국 표준시 설정 KST = timezone(timedelta(hours=9)) -# 성능 최적화 설정 -DEFAULT_BATCH_SIZE = 500 -DEFAULT_MAX_WORKERS = 6 -DEFAULT_COMPOSITE_SIZE = 500 +# 성능 최적화 설정 (오픈서치 스펙 기반 최적화) +DEFAULT_BATCH_SIZE = 2000 +DEFAULT_MAX_WORKERS = 16 +DEFAULT_COMPOSITE_SIZE = 2000 DEFAULT_TIMEOUT = 180 SCROLL_TIMEOUT = "5m" SESSION_GAP_MINUTES = 5 @@ -680,9 +680,25 @@ def get_new_user_cohort_optimized( auth_id_count = sum(1 for uid in new_user_map if new_user_map[uid]["auth_id"] is not None) logger.info(f"auth.id 수집 완료: {auth_id_count}/{len(new_user_map)}명") - # Step 3: login_comp 인덱스에서 추가 정보 수집 + # Step 3: 모든 create_uid 유저를 cohort에 추가하고, 추가 정보 수집 + logger.info("모든 신규 유저를 cohort에 추가하고 추가 정보 수집 중...") + + # 모든 create_uid 유저를 cohort에 먼저 추가 + for uid in uid_list: + cohort[uid] = { + 'auth_id': new_user_map[uid]["auth_id"] or 'N/A', + 'create_time_utc': new_user_map[uid]["create_time"], + 'create_time_kst': format_kst_time(new_user_map[uid]["create_time"]), + 'create_time_dt': datetime.fromisoformat(new_user_map[uid]["create_time"].replace('Z', '+00:00')), + 'language': 'N/A', + 'device': 'N/A', + 'nickname': 'N/A' + } + total_users += 1 + + # login_comp 인덱스에서 추가 정보 수집 (1차 우선) logger.info("login_comp 인덱스에서 추가 정보 수집 중...") - # 유저 청크 단위로 처리 + login_comp_collected = set() for i in range(0, len(uid_list), chunk_size): chunk_uids = uid_list[i:i+chunk_size] @@ -703,7 +719,6 @@ def get_new_user_cohort_optimized( "size": chunk_size }, "aggs": { - "first_login": {"min": {"field": "@timestamp"}}, "user_info": { "top_hits": { "size": 1, @@ -734,27 +749,84 @@ def get_new_user_cohort_optimized( for bucket in response["aggregations"]["users"]["buckets"]: uid = bucket["key"] - first_login_utc = bucket["first_login"]["value_as_string"] user_hit = bucket["user_info"]["hits"]["hits"][0]["_source"] if bucket["user_info"]["hits"]["hits"] else {} latest_info_hit = bucket["latest_info"]["hits"]["hits"][0]["_source"] if bucket["latest_info"]["hits"]["hits"] else {} - # create_uid 정보와 병합 (first_login 제거, create_time으로 통합) - # auth_id는 heartbeat에서 수집되었거나 N/A - cohort[uid] = { - 'auth_id': new_user_map[uid]["auth_id"] or 'N/A', - 'create_time_utc': new_user_map[uid]["create_time"], - 'create_time_kst': format_kst_time(new_user_map[uid]["create_time"]), - 'create_time_dt': datetime.fromisoformat(new_user_map[uid]["create_time"].replace('Z', '+00:00')), - 'language': latest_info_hit.get('body', {}).get('language', 'N/A'), - 'device': user_hit.get('body', {}).get('device_mod', 'N/A'), - 'nickname': latest_info_hit.get('body', {}).get('nickname') or user_hit.get('body', {}).get('nickname', 'N/A') - } - total_users += 1 + # 기존 cohort 정보 업데이트 + if uid in cohort: + cohort[uid]['language'] = latest_info_hit.get('body', {}).get('language', 'N/A') + cohort[uid]['device'] = user_hit.get('body', {}).get('device_mod', 'N/A') + cohort[uid]['nickname'] = latest_info_hit.get('body', {}).get('nickname') or user_hit.get('body', {}).get('nickname', 'N/A') + login_comp_collected.add(uid) except Exception as e: logger.error(f"login_comp 정보 수집 중 오류: {e}") + logger.info(f"login_comp에서 {len(login_comp_collected)}명의 추가 정보 수집 완료") + + # Step 4: log_return_to_lobby 인덱스에서 차선 정보 수집 (nickname만) + missing_uids = [uid for uid in uid_list if uid not in login_comp_collected and cohort[uid]['nickname'] == 'N/A'] + + if missing_uids: + logger.info(f"log_return_to_lobby 인덱스에서 {len(missing_uids)}명의 차선 정보 수집 중...") + lobby_collected = set() + + for i in range(0, len(missing_uids), chunk_size): + chunk_uids = missing_uids[i:i+chunk_size] + + lobby_query = { + "size": 0, + "query": { + "bool": { + "filter": [ + {"terms": {"uid.keyword": chunk_uids}} + ] + } + }, + "aggs": { + "users": { + "terms": { + "field": "uid.keyword", + "size": chunk_size + }, + "aggs": { + "nickname_info": { + "top_hits": { + "size": 1, + "sort": [{"@timestamp": {"order": "desc"}}], + "_source": ["body.nickname"] + } + } + } + } + } + } + + try: + response = exponential_backoff_retry( + client.search, + index="ds-logs-live-log_return_to_lobby", + body=lobby_query, + request_timeout=DEFAULT_TIMEOUT, + track_total_hits=False + ) + + for bucket in response["aggregations"]["users"]["buckets"]: + uid = bucket["key"] + + nickname_hit = bucket["nickname_info"]["hits"]["hits"][0]["_source"] if bucket["nickname_info"]["hits"]["hits"] else {} + + # nickname만 업데이트 (다른 정보는 login_comp가 우선) + if uid in cohort and cohort[uid]['nickname'] == 'N/A': + cohort[uid]['nickname'] = nickname_hit.get('body', {}).get('nickname', 'N/A') + lobby_collected.add(uid) + + except Exception as e: + logger.error(f"log_return_to_lobby 정보 수집 중 오류: {e}") + + logger.info(f"log_return_to_lobby에서 {len(lobby_collected)}명의 닉네임 수집 완료") + logger.info(f"1단계 완료: 총 {total_users}명의 신규 유저 코호트 확정 (create_uid 기준)") logger.info("=" * 80) return cohort