Как обработать rdd?

Question

[Denis]

у меня есть rdd, который выглядит так:

my_rdd.take(10)

[['movieId', 'title', 'genres'],
 ['1', 'Toy Story (1995)', 'Adventure|Animation|Children|Comedy|Fantasy'],
 ['2', 'Jumanji (1995)', 'Adventure|Children|Fantasy'],
 ['3', 'Grumpier Old Men (1995)', 'Comedy|Romance'],
 ['4', 'Waiting to Exhale (1995)', 'Comedy|Drama|Romance'],
 ['5', 'Father of the Bride Part II (1995)', 'Comedy'],
 ['6', 'Heat (1995)', 'Action|Crime|Thriller'],
 ['7', 'Sabrina (1995)', 'Comedy|Romance'],
 ['8', 'Tom and Huck (1995)', 'Adventure|Children'],
 ['9', 'Sudden Death (1995)', 'Action']]

и у меня есть переменные, по которым мне нужно отфильтровать этот my_rdd:

arg_genres = 'Action' # фильтр по жанру
year_to = 2012 # фильтр по году 
year_from = 2005 # фильтр по году 
regexp = 'Toy Story' # фильтр по названию фильма

но я не понимаю как мне получить новый rdd в котором будут содержаться фильмы те фильмы которые удовлетворяют переменным для фильтрации, которые я указал выше.
Каким способом можно получить новый rdd с отфильтрованными фильмами?

Comments

[Алан Гибизов]

Нууу… по этим параметрам будет пустой список ответом.
А отфильтровать надо каким образом? Надо сформировать запрос правильно, чтоб сразу получить ответ, или надо вот этот полученный список списков дополнительно фильтровать?

[Denis]

Алан Гибизов,
вот что у меня есть, пишу в zeppeline notebook

ARG_GENRES = '' # по дефолту равен '',  но может принимать такие значение "Action|Comedy" или просто "Adventure"
YEAR_TO = 2009  # по дефолту равен 2030 чтобы точно все фильмы подошли 
YEAR_FROM = 2000 # по дефолту равен 1800 чтобы точно все фильмы подошли
REGEXP = 'The'  # по дефолту равен '' чтобы все значения подошли
N = 10  #? фильтр для колличества вывода результата

def get_line_from_string(bed_line):
    result = []
    b = bed_line.split(',', maxsplit=1)
    c = b[1].rsplit(',', maxsplit=1)
    if '"' in c[0]:
        c[0] = c[0][1:-1]
    line = [b[0], c[0], c[1]]
    return line

def filter_by_regexp(string):
    pattern = REGEXP
    if re.search(pattern, string):
        return True

def filter_by_year(string):
    pattern = r'\(\d{4}\)'
    if re.search(pattern, string):
        year = int(re.search(r'\(\d{4}\)', string).group(0)[1:-1])
        if YEAR_FROM <= year <= YEAR_TO:
            return True

rdd = sc.textFile('hdfs:///movies.csv').map(lambda x: get_line_from_string(x))
rdd = rdd.filter(lambda list: filter_by_regexp(list[1]) and filter_by_year(list[1]))
rdd.take(5)

[['2769', 'Yards, The (2000)', 'Crime|Drama'],
['3275', 'Boondock Saints, The (2000)', 'Action|Crime|Drama|Thriller'],
['3285', 'Beach, The (2000)', 'Adventure|Drama'],
['3287', 'Tigger Movie, The (2000)', 'Animation|Children'],
['3301', 'Whole Nine Yards, The (2000)', 'Comedy|Crime']]

def get_list_from_rating_line(line):
    _,movieId, rating,_ = line.split(',')
    
    return [movieId, rating]

rating_rdd = sc.textFile('hdfs:///ratings.csv')
rating_rdd = rating_rdd.map(lambda x: get_list_from_rating_line(x))
rating_rdd.take(5)

[['movieId', 'rating'],
['1', '4.0'],
['3', '4.0'],
['6', '4.0'],
['47', '5.0']]

архив с файлами можно скачать по этой ссылке: https://files.grouplens.org/datasets/movielens/ml-...

то есть, у меня сейчас есть rdd это рдд c фильмами, и у меня есть rating_rdd это рдд с рейтингами по фильмам,

в будующем, я я обьеденю эти рдд в один, но сейчас основной проблемой является то, что мне нужно разбить значения rdd по жанрам, то есть, это означает,
rdd.take(3) выгдлядит так:
[['2769', 'Yards, The (2000)', 'Crime|Drama'],
['3275', 'Boondock Saints, The (2000)', 'Action|Crime|Drama|Thriller'],
['3285', 'Beach, The (2000)', 'Adventure|Drama']]

а мне нужно, чтобы он выглядел так:
rdd.take(8)

[['2769', 'Yards, The (2000)', 'Crime'],
['2769', 'Yards, The (2000)', 'Drama'],
['3275', 'Boondock Saints, The (2000)', 'Action'],
['3275', 'Boondock Saints, The (2000)', 'Crime'],
['3275', 'Boondock Saints, The (2000)', 'Drama'],
['3275', 'Boondock Saints, The (2000)', 'Thriller'],
['3285', 'Beach, The (2000)', 'Adventure'],
['3285', 'Beach, The (2000)', 'Drama']]

Я не могу разить массивы по жанрам, просто rdd.map(lambda x: какая-то_функция_для_разбиения_по_жанрам(x)) не работает, по скольку при таком условии мне нужно делать yield значений, а не return, а с yield рдд почему то не работаю

Фильтровать фильмы нужно просто меняя аргументы для фильтарции, то есть просто менять ARG_GENRES YEAR_TO YEAR_FROM REGEXP N