Apache Pig - Как получить количество совпадающих элементов между несколькими мешками?

Question

Я новый пользователь Apache Pig, и у меня есть проблема для решения.

Я пытаюсь создать небольшую поисковую систему с апачей. Идея проста: у меня есть файл, который является конкатенацией нескольких документов (по одному документу на строку). Вот пример с тремя документами:

1,word1 word4 word2 word1 
2,word2 word6 word1 word5 word3 
3,word1 word3 word4 word5 

Затем я создаю мешок слов для каждого из документов, используя следующие строки кода:

docs = LOAD '$documents' USING PigStorage(',') AS (id:int, line:chararray); 
B = FOREACH docs GENERATE line; 
C = FOREACH B GENERATE TOKENIZE(line) as gu; 

Затем я удалить дубликаты записей на пакетах:

filtered = FOREACH C { 
    uniq = DISTINCT gu; 
    GENERATE uniq; 
} 

Вот результаты этого кода:

DUMP filtered; 

({(word1), (word4), (word2)}) 
({(word2), (word6), (word1), (word5), (word3)}) 
({(word1), (word3), (word4), (word5)}) 

Итак, у меня есть мешок слов на документ, как я и хотел.

Теперь давайте рассмотрим запрос пользователя в виде файла:

word2 word7 word5 

Я преобразовать запрос в мешок слов:

query = LOAD '$query' AS (line_query:chararray); 
bag_query = FOREACH query GENERATE TOKENIZE(line_query) AS quer; 

DUMP bag_query; 

Вот результаты:

({(word2), (word7), (word5)}) 

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

1 
2 
1 

Я пытался сделать JOIN между мешками, но это не сработало.

Не могли бы вы помочь мне, пожалуйста?

спасибо.

Answer 1

Если вы не используете какой-либо из UDF, то это может быть сделано путем поворота мешков и перехода на весь стиль SQL.

docs = LOAD '/input/search.dat' USING PigStorage(',') AS (id:int, line:chararray); 
C = FOREACH docs GENERATE id, TOKENIZE(line) as gu; 
pivoted = FOREACH C { 
    uniq = DISTINCT gu; 
     GENERATE id, FLATTEN(uniq) as word; 
}; 
filtered = FILTER pivoted BY word MATCHES '(word2|word7|word5)'; 
--dump filtered; 
count_id_matched = FOREACH (GROUP filtered BY id) GENERATE group as id, COUNT(filtered) as count; 

dump count_id_matched; 

count_word_matched_in_docs = FOREACH (GROUP filtered BY word) GENERATE group as word, COUNT(filtered) as count; 

dump count_word_matched_in_docs; 

Answer 2

Попробуйте использовать SetIntersect (Datafu UDF - https://github.com/linkedin/datafu) и РАЗМЕР, чтобы получить количество элементов в сумке результатов.

Answer 3

Как указал SNeumann, вы можете использовать SetIntersect DataFu для своего примера.

Строительства своего примера, учитывая эти документы:

1,word1 word4 word2 word1 
2,word2 word6 word1 word5 word3 word7 
3,word1 word3 word4 word5 

И учитывая этот вопрос:

word2 word7 word5 

Тогда этот код дает вам то, что вы хотите:

define SetIntersect datafu.pig.sets.SetIntersect(); 

docs = LOAD 'docs' USING PigStorage(',') AS (id:int, line:chararray); 
B = FOREACH docs GENERATE id, line; 
C = FOREACH B GENERATE id, TOKENIZE(line) as gu; 

filtered = FOREACH C { 
    uniq = DISTINCT gu; 
    GENERATE id, uniq; 
} 

query = LOAD 'query' AS (line_query:chararray); 
bag_query = FOREACH query GENERATE TOKENIZE(line_query) AS query; 
-- sort the bag of tokens, since SetIntersect requires it 
bag_query = FOREACH bag_query { 
    query_sorted = ORDER query BY token; 
    GENERATE query_sorted; 
} 

result = FOREACH filtered { 
    -- sort the tokens, since SetIntersect requires it 
    tokens_sorted = ORDER uniq BY token; 
    GENERATE id, 
      SIZE(SetIntersect(tokens_sorted,bag_query.query_sorted)) as cnt; 
} 

DUMP result; 

Значение для результат:

(1,1) 
(2,3) 
(3,1) 

Вот полностью рабочий пример, который вы можете вставить в юнит-тесты DataFu для SetIntersect расположенных here:

/** 
register $JAR_PATH 

define SetIntersect datafu.pig.sets.SetIntersect(); 

docs = LOAD 'docs' USING PigStorage(',') AS (id:int, line:chararray); 
B = FOREACH docs GENERATE id, line; 
C = FOREACH B GENERATE id, TOKENIZE(line) as gu; 

filtered = FOREACH C { 
    uniq = DISTINCT gu; 
    GENERATE id, uniq; 
} 

query = LOAD 'query' AS (line_query:chararray); 
bag_query = FOREACH query GENERATE TOKENIZE(line_query) AS query; 
-- sort the bag of tokens, since SetIntersect requires it 
bag_query = FOREACH bag_query { 
    query_sorted = ORDER query BY token; 
    GENERATE query_sorted; 
} 

result = FOREACH filtered { 
    -- sort the tokens, since SetIntersect requires it 
    tokens_sorted = ORDER uniq BY token; 
    GENERATE id, 
      SIZE(SetIntersect(tokens_sorted,bag_query.query_sorted)) as cnt; 
} 

DUMP result; 

*/ 
@Multiline 
private String setIntersectTestExample; 

@Test 
public void setIntersectTestExample() throws Exception 
{  
    PigTest test = createPigTestFromString(setIntersectTestExample);  

    writeLinesToFile("docs", 
        "1,word1 word4 word2 word1", 
        "2,word2 word6 word1 word5 word3 word7", 
        "3,word1 word3 word4 word5"); 

    writeLinesToFile("query", 
        "word2 word7 word5"); 

    test.runScript(); 

    super.getLinesForAlias(test, "filtered"); 
    super.getLinesForAlias(test, "query"); 
    super.getLinesForAlias(test, "result"); 
} 

Если у вас есть какие-либо другие подобные прецеденты, я хотел бы услышать их :) Мы всегда стремясь внести более полезные UDF в DataFu.