Pyspark: rdd в json объекте

Question

Это домашнее задание

hw2-files-10mb.txt - это данные json

rdd = sc.textFile('./hw2-files-10mb.txt')
rdd = rdd.map(json.loads)

Output of rdd.take(1):

[{'created_at': 'Tue Feb 23 17:42:31 +0000 2016',
  'user': {'id': 470520068,
    'id_str': '470520068',
    'name': 'Marni Halasa',
    'screen_name': 'MarniHalasa1',
    'location': 'NYC',
....]

Вопрос:

1) Как выбрать 'id_str' в 'user 'в режиме rdd?

Я попробовал метод для отображения rdd в func, который возвращает field_list = [' user.id_str '], но он не работает.

field_list = ['user.id_str', 'text']
def f(x):
d = {}
for k in x:
    if k in field_list:
        d[k] = x[k]
return d

rdd1 = rdd.map(f)

Output:
[{'text': "I'm voting 4 #BernieSanders bc he doesn't ride a CAPITALIST PIG adorned w/ #GoldmanSachs $. SYSTEM RIGGED CLASS WAR"}]

2) Как убрать твит, в котором не было создано поле create_at?

Мои коды ниже не работают.Он возвращает ошибку, когда я использую rdd.count ()

rdd = rdd.filter(lambda row: row['created_at'] is not None)

Мой ожидаемый вывод:

1) Удалите все битые твиты (поле 'create_at' пусто)

2) пара СДР из (user_id, text), где user_id - поле данных 'id_str' в пользовательском словаре.

Спасибо.

Answer 1

Сначала решите точку 2), так как после извлечения 'user.id_str' и 'text' не останется поля created_at для фильтрации.

2) Просто проверьте, если 'created_at'подарок.Кроме того, вы можете выполнять все операции за один раз.

rdd = sc.textFile('./hw2-files-10mb.txt')\
            .map(json.loads)\
            .filter(lambda row: 'created_at' in row.keys())

1) Вам не нужна функция для извлечения значений из объекта json.

rdd = rdd.map(lambda row:(row['user']['id_str'], 'text'))
print rdd.take(1)