Как я могу пропустить n строк двоичного стандартного ввода, используя Python?

Question

Я передаю двоичные данные в скрипт Python на кластере Hadoop с помощью интерфейса командной строки Hadoop.Двоичные данные имеют терминаторы, которые определяют, где начинаются новые документы.Записи сортируются по уникальному идентификатору, который начинается с 1000000001 и увеличивается на 1.

Я пытаюсь сохранить данные только для подмножества этих идентификаторов, которые есть в словаре.

Мой текущий процесс состоит в том, чтобы выбрать данные из CLI, используя:

hadoop select "Database" "Collection" | cut -d$'\t' -f2 | python script.py

и обработать их в script.py, который выглядит следующим образом:

import json
import sys

member_mapping = json.load(open('member_mapping.json'))

output = []

for line in sys.stdin:
    person = json.loads(line)
    if member_mapping.get(person['personId']):
        output.append({person['personId']: person})
    if len(output) == len(member_mapping):
        break

Проблема в том, что есть 6,5M идентификаторов в этих двоичных данных, и сканирование занимает почти 2 часа.Я знаю идентификаторы min () и max () в моем словаре, и вы можете видеть в моем коде, что я остановился раньше, когда сохранил n документов, где n - длина моего файла сопоставления.

Я хочусделайте этот процесс более эффективным, пропустив как можно больше операций чтения.Если идентификатор начинается с 1000000001, а первый идентификатор, который я хочу сохранить, это 1000010001, могу ли я просто пропустить 10000 строк?

Из-за проблем с системой я не могу использовать искруили любые другие инструменты, которые могут улучшить этот процесс, поэтому мне нужно пока придерживаться решений, использующих Python и Hadoop CLI.

Answer 1

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

for lineNum, line in enumerate(sys.stdin):
    if(lineNum < 10000):
         continue
    person = json.loads(line)
    if member_mapping.get(person['personId']):
        output.append({person['personId']: person})
    if len(output) == len(member_mapping):
        break