Лучший способ использовать Apache Hive для сессии моих данных журнала?

Question

Есть ли лучший способ использовать Apache Hive для сессии моих данных журнала?Я не уверен, что делаю это ниже, оптимальным образом:

Данные журнала хранятся в файлах последовательности;отдельная запись в журнале является строкой JSON;Например:

{"source": {"api_key": "app_key_1", "user_id": "user0"}, "events": [{"timestamp": 1330988326, "event_type": "high_score", "event_params": {"score": "1123", "level": "9"}}, {"timestamp": 1330987183, "event_type": "some_event_0", "event_params": {"some_param_00": "val", "some_param_01": 100}}, {"timestamp": 1330987775, "event_type": "some_event_1", "event_params": {"some_param_11": 100, "some_param_10": "val"}}]}

Отформатировано, это выглядит так:

{'source': {'api_key': 'app_key_1', 'user_id': 'user0'},
 'events': [{'event_params': {'level': '9', 'score': '1123'},
             'event_type': 'high_score',
             'timestamp': 1330988326},
            {'event_params': {'some_param_00': 'val', 'some_param_01': 100},
             'event_type': 'some_event_0',
             'timestamp': 1330987183},
            {'event_params': {'some_param_10': 'val', 'some_param_11': 100},
             'event_type': 'some_event_1',
             'timestamp': 1330987775}]
}

'source' содержит некоторую информацию (user_id и api_key) об источнике содержащихся событийв «событиях»;'events' содержит список событий, сгенерированных источником;каждое событие имеет «event_params», «event_type» и «timestamp» (timestamp - это метка времени Unix в GMT).Обратите внимание, что временные метки в одной записи журнала и между записями журнала могут быть не в порядке.

Обратите внимание, что я ограничен таким образом, что не могу изменить формат журнала, изначально не могу записать данные в отдельные файлы, которыесекционирован (хотя я мог бы использовать Hive для этого после регистрации данных) и т. д.

В конце я бы хотел таблицу сессий, где сессия связана с приложением (api_k) ипользователь, и имеет время начала и продолжительность сеанса (или время окончания);сеансы разделяются, когда для данного приложения и пользователя между событиями возникает разрыв в 30 и более минут.

Мое решение делает следующее (сценарий Hive и сценарий преобразования Python приведены ниже; не похоже на этобыло бы полезно показать источник SerDe, но дайте мне знать, если это так):

[1] загрузить данные в log_entry_tmp, в денормализованном формате

[2] взорвать данныев log_entry, так что, например, вышеупомянутая одиночная запись теперь будет иметь несколько записей:

{"source_api_key":"app_key_1","source_user_id":"user0","event_type":"high_score","event_params":{"score":"1123","level":"9"},"event_timestamp":1330988326}
{"source_api_key":"app_key_1","source_user_id":"user0","event_type":"some_event_0","event_params":{"some_param_00":"val","some_param_01":"100"},"event_timestamp":1330987183}
{"source_api_key":"app_key_1","source_user_id":"user0","event_type":"some_event_1","event_params":{"some_param_11":"100","some_param_10":"val"},"event_timestamp":1330987775}

[3] преобразовывать и записывать данные в session_info_0, где каждая запись содержит события app_id, user_id и timestamp

[4] преобразовывать и записывать данные в session_info_1, где записи упорядочены по app_id, user_id, event_timestamp;и каждая запись содержит идентификатор_сессии;скрипт преобразования python находит разбиения и группирует данные в сеансы

[5] преобразует и записывает окончательные данные сеанса в session_info_2;приложение сеансов + пользователь, время начала и продолжительность в секундах

---

[скрипт Hive]

drop table if exists app_info;
create external table app_info ( app_id int, app_name string, api_k string )
location '${WORK}/hive_tables/app_info';

add jar ../build/our-serdes.jar;

-- [1] load the data into log_entry_tmp, in a denormalized format

drop table if exists log_entry_tmp;
create external table log_entry_tmp
row format serde 'com.company.TestLogSerde'
location '${WORK}/hive_tables/test_logs';

drop table if exists log_entry;
create table log_entry (
    entry struct<source_api_key:string,
                 source_user_id:string,
                 event_type:string,
                 event_params:map<string,string>,
                 event_timestamp:bigint>);

-- [2] explode the data into log_entry

insert overwrite table log_entry
select explode (trans0_list) t
from log_entry_tmp;

drop table if exists session_info_0;
create table session_info_0 (
    app_id string,
    user_id string,
    event_timestamp bigint
);

-- [3] transform and write data into session_info_0, where each entry contains events' app_id, user_id, and timestamp

insert overwrite table session_info_0
select ai.app_id, le.entry.source_user_id, le.entry.event_timestamp
from log_entry le
join app_info ai on (le.entry.source_api_key = ai.api_k);

add file ./TestLogTrans.py;

drop table if exists session_info_1;
create table session_info_1 (
    session_id string,
    app_id string,
    user_id string,
    event_timestamp bigint,
    session_start_datetime string,
    session_start_timestamp bigint,
    gap_secs int
);

-- [4] tranform and write data into session_info_1, where entries are ordered by app_id, user_id, event_timestamp ; and each entry contains a session_id ; the python tranform script finds the splits, and groups the data into sessions

insert overwrite table session_info_1
select
    transform (t.app_id, t.user_id, t.event_timestamp)
        using './TestLogTrans.py'
        as (session_id, app_id, user_id, event_timestamp, session_start_datetime, session_start_timestamp, gap_secs)
from
    (select app_id as app_id, user_id as user_id, event_timestamp as event_timestamp from session_info_0 order by app_id, user_id, event_timestamp ) t;

drop table if exists session_info_2;
create table session_info_2 (
    session_id string,
    app_id string,
    user_id string,
    session_start_datetime string,
    session_start_timestamp bigint,
    len_secs int
);

-- [5] transform and write final session data to session_info_2 ; the sessions' app + user, start time, and length in seconds

insert overwrite table session_info_2
select session_id, app_id, user_id, session_start_datetime, session_start_timestamp, sum(gap_secs)
from session_info_1
group by session_id, app_id, user_id, session_start_datetime, session_start_timestamp;

---

[TestLogTrans.py]

#!/usr/bin/python

import sys, time

def buildDateTime(ts):

    return time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(ts))

curGroup = None
prevGroup = None
curSessionStartTimestamp = None
curSessionStartDatetime = None
prevTimestamp = None

for line in sys.stdin.readlines():

    fields = line.split('\t')
    if len(fields) != 3:
        raise Exception('fields = %s', fields)

    app_id = fields[0]
    user_id = fields[1]
    event_timestamp = int(fields[2].strip())

    curGroup = '%s-%s' % (app_id, user_id)
    curTimestamp = event_timestamp

    if prevGroup == None:
        prevGroup = curGroup
        curSessionStartTimestamp = curTimestamp
        curSessionStartDatetime = buildDateTime(curSessionStartTimestamp)
        prevTimestamp = curTimestamp

    isNewGroup = (curGroup != prevGroup)

    gapSecs = 0 if isNewGroup else (curTimestamp - prevTimestamp)

    isSessionSplit = (gapSecs >= 1800)

    if isNewGroup or isSessionSplit:
        curSessionStartTimestamp = curTimestamp
        curSessionStartDatetime = buildDateTime(curSessionStartTimestamp)

    session_id = '%s-%s-%d' % (app_id, user_id, curSessionStartTimestamp)

    print '%s\t%s\t%s\t%d\t%s\t%d\t%d' % (session_id, app_id, user_id, curTimestamp, curSessionStartDatetime, curSessionStartTimestamp, gapSecs)

    prevGroup = curGroup
    prevTimestamp = curTimestamp

Answer 1

Я думаю, что вы могли бы легко отказаться от шага 3 и поместить этот запрос, который вы там используете, в качестве подзапроса к предложению from на шаге 4. Физикализация этого преобразования, похоже, ничего вам не даст.

В противном случае я думаю, что для того, что вы пытаетесь достичь здесь, это кажется разумным подходом.

Потенциально шаг 2, который вы могли бы достичь, используя собственный преобразователь, передавая вывод в шаг 4 в качестве обычного редуктора (с шагом3 встроено как подзапрос).Это сократит ваши рабочие места в Mapreduce на 1 и, следовательно, может дать вам значительную экономию времени.