Spark : RDD 맵에서 반복자 람다 함수 사용하기 (

Question

)

나는 Spark에로드하는 HDFS에서 간단한 데이터 세트를 가지고 있습니다. 이것은 다음과 같습니다 :

1 1 1 1 1 1 1 
1 1 1 1 1 1 1 
1 1 1 1 1 1 1 
... 

기본적으로 매트릭스입니다. 나는 그룹 행렬 행을 필요로 뭔가를 구현하기 위해 노력하고있어, 그래서 나는과 같이 모든 행에 대한 고유 키를 추가하기 위해 노력하고있어 : 전역 변수를 설정하고 쓰기 :

(1, [1 1 1 1 1 ... ]) 
(2, [1 1 1 1 1 ... ]) 
(3, [1 1 1 1 1 ... ]) 
... 

내가 다소 순진 뭔가를 시도 lambda 함수를 사용하여 전역 변수를 반복 할 수 있습니다.

# initialize global index 
global global_index 
global_index = 0 

# function to generate keys 
def generateKeys(x): 
    global_index+=1 
    return (global_index,x) 

# read in data and operate on it 
data = sc.textFile("/data.txt") 

...some preprocessing... 

data.map(generateKeys) 

그리고 전역 변수의 존재를 인식하지 못하는 것처럼 보입니다.

쉬운 방법이 있나요?

감사합니다, 잭

Answer 1

>>> lsts = [ 
...  [1, 1, 1, 1, 1, 1], 
...  [1, 1, 1, 1, 1, 1], 
...  [1, 1, 1, 1, 1, 1], 
...  [1, 1, 1, 1, 1, 1], 
...  [1, 1, 1, 1, 1, 1], 
...  [1, 1, 1, 1, 1, 1], 
...  [1, 1, 1, 1, 1, 2], 
...  [1, 1, 1, 2, 1, 2] 
...  ] 
... 
>>> list(enumerate(lsts)) 
[(0, [1, 1, 1, 1, 1, 1]), 
(1, [1, 1, 1, 1, 1, 1]), 
(2, [1, 1, 1, 1, 1, 1]), 
(3, [1, 1, 1, 1, 1, 1]), 
(4, [1, 1, 1, 1, 1, 1]), 
(5, [1, 1, 1, 1, 1, 1]), 
(6, [1, 1, 1, 1, 1, 2]), 
(7, [1, 1, 1, 2, 1, 2])] 

enumerate 값이 0 이외의 번호 매기기를 시작하려면 (index, original_item)

와 반복 가능한 수율 튜플의 각 항목에 대해 고유 인덱스를 생성, 시작 값을 전달 제 2 파라미터로서 enumerate으로 설정한다.

>>> list(enumerate(lsts, 1)) 
[(1, [1, 1, 1, 1, 1, 1]), 
(2, [1, 1, 1, 1, 1, 1]), 
(3, [1, 1, 1, 1, 1, 1]), 
(4, [1, 1, 1, 1, 1, 1]), 
(5, [1, 1, 1, 1, 1, 1]), 
(6, [1, 1, 1, 1, 1, 1]), 
(7, [1, 1, 1, 1, 1, 2]), 
(8, [1, 1, 1, 2, 1, 2])] 

참고 list이 반복자와 목록을 반환하지 않는 함수입니다 enumerate에서 실제 값을 얻기 위해 사용된다.

대안 : 전 세계적으로 사용 가능한 ID를 지정 기

enumerate를 사용하기 쉽습니다,하지만 당신은 코드의 diferrent 조각 ID를 assing 할 필요가 있다면, 그것은 어렵거나 불가능하게 될 것입니다. 이 경우 전 세계적으로 사용 가능한 발전기 ( OP의 기안자)가 될 수 있습니다.

itertools은 우리의 필요를 제공 할 수 count 제공 :

>>> from itertools import count 
>>> idgen = count() 

이제 우리는 (세계적으로 사용 가능) 고유 ID를 생성 할 idgen 발전기 준비.

우리는 기능 prid (프린트 ID)하여 테스트 할 수

>>> def prid(): 
...  id = idgen.next() 
...  print id 
... 
>>> prid() 
0 
>>> prid() 
1 
>>> prid() 
2 
>>> prid() 
3 

우리가 값 목록에서 테스트 할 수 있습니다 작동과 같이 실제 기능을

>>> lst = ['100', '101', '102', '103', '104', '105', '106', '107', '108', '109'] 

및 정의하는 경우 값으로 불리는 값은 튜플을 반환합니다. (id, value)

>>> def assignId(val): 
...  return (idgen.next(), val) 
... 

주 idgen을 전역으로 선언 할 필요가 없으므로 값을 변경하지 않습니다 (idgen은 호출시 내부 상태 만 변경하지만 동일한 생성기로 유지됩니다).

>>> assignId("ahahah") 
(4, 'ahahah') 

를하고 목록에 그것을 시도 :

테스트, 그것은 작동하는 경우

>>> map(assignId, lst) 
[(5, '100'), 
(6, '101'), 
(7, '102'), 
(8, '103'), 
(9, '104'), 
(10, '105'), 
(11, '106'), 
(12, '107'), 
(13, '108'), 
(14, '109')] 

주요 diferrence enumerate에 솔루션입니다, 우리는 어디 코드에 의해 식별자를 하나씩 할당 할 수 있습니다 모든 것을 처리하지 않고 모든 처리 enumerate.

>>> assignId("lonely line") 
(15, 'lonely line') 

Answer 2

는 dataRdd.zipWithIndex을 시도하고 첫번째 인덱스를 갖는 것은 필수입니다 경우 결국 결과 튜플을 교환합니다.