inputFile = r'C:\Data\sample.txt'
rddData = sparkContext.textFile(inputFile)
rddData.flatMap(lambda x: x.split()).map(lambda x: (x, 1)).reduceByKey(lambda a, b: a+b)

2. RDD 생성

(1) 컬렉션 데이터에서 RDD 생성

numbers = [10, 20, 30, 40, 50]
baseRDD1 = sc.parallelize(numbers)
baseRDD2 = sc.parallelize(numbers, 3)
baseRDD2.count()

반복(Iteration): list, tuple, dict, set, str 등의 자료형에 for...in... 문을 사용하여 순차적으로 데이터를 추출하여 처리하는 과정을 의미한다.

반복 가능 객체(Iterable): for...in... 문으로 순차적으로 데이터를 읽어 사용할 수 있는 객체를 말한다.

(2) 외부 데이터 소스에서 RDD 생성

textFile 함수를 활용하면 외부 데이터 소스를 RDD 객체로 변환할 수 있다. 파일의 각 줄이 RDD의 하나의 요소가 되며, 데이터 소스의 경로를 인자로 전달해야 한다.

localFile = sc.textFile("file:///home/user/data/sample.txt")
hdfsFile = sc.textFile("hdfs://namenode:9000/user/test/data")
localFile.count()

3. RDD 연산자

RDD에서 수행하는 Operation(작업)은 변환(Transformation)과 동작(Action)으로 구분된다.

RDD는 MR(MapReduce)보다 훨씬 다양한 연산을 지원한다. Map, Reduce뿐만 아니라 filter, sort, join, save, count 등의 작업을 포함하므로 Spark는 더 복잡한 작업을 쉽게 처리할 수 있다.

(1) 변환 연산 map

지정한 함수를 RDD의 각 파티션에 속한 모든 요소에 적용한다.

rdd = sc.parallelize(["x", "y", "z"])
rdd.map(lambda elem: (elem, 1)).collect()
결과:
[('x', 1), ('y', 1), ('z', 1)]

map은 RDD의 각 요소를 순차적으로 추출하여 lambda 함수의 elem 파라미터로 전달한 후, tuple (elem, 1)을 생성하는 작업을 수행한다. collect()는 각 메모리에 분산된 결과를 드라이버로 반환하며, RDD의 모든 요소를 포함하는 리스트를 돌려준다.

(2) 변환 연산 flatMap

먼저 map 함수를 RDD의 모든 요소에 적용한 후, 결과를 평탄화(각 요소를 개별적으로 추출)하여 새로운 RDD를 반환한다.

rdd = sc.parallelize([5, 6, 7])
rdd.flatMap(lambda x: [(x, x*2), (x, x*2)]).collect()
[(5, 10), (5, 10), (6, 12), (6, 12), (7, 14), (7, 14)]

만약 flatMap 대신 map을 사용하면:
[[(5, 10), (5, 10)], [(6, 12), (6, 12)], [(7, 14), (7, 14)]]

(3) 변환 연산 filter

지정한 함수를 필터 조건으로 사용하여 RDD의 모든 파티션에서 조건을 만족하는 요소만 선별한다.

rdd = sc.parallelize([1, 2, 3, 4, 5])
rdd.filter(lambda x: x > 3).collect()
[4, 5]

(4) 변환 연산 union

두 RDD의 합집합을 구하여 새로운 RDD를 반환한다. 이 연산은 중복을 제거하지 않는다.

rdd1 = sc.parallelize([1, 2, 3, 4])
rdd2 = sc.parallelize([3, 4, 5, 6])
rdd1.union(rdd2).collect()
[1, 2, 3, 4, 3, 4, 5, 6]

(5) 변환 연산 intersection

두 RDD의 교집합을 구하여 새로운 RDD를 반환한다.

rdd1 = sc.parallelize([1, 2, 3, 4])
rdd2 = sc.parallelize([3, 4, 5, 6])
rdd1.intersection(rdd2).collect()
[3, 4]

(6) 변환 연산 distinct

리스트를 집합(set)으로 변환하듯이, RDD에서 중복된 요소를 제거한 서로 다른 요소만 포함하는 새로운 RDD를 반환한다.

sourceRDD = sc.parallelize([3, 4, 5, 6, 3, 4])
sourceRDD.distinct().collect()
[3, 4, 5, 6]

(7) 변환 연산 sortBy

지정한 키를 기준으로 RDD의 요소를 정렬하여 새로운 RDD를 반환한다. 기본값은 오름차순 정렬이며, 내림차순으로 정렬하려면 ascending=False 파라미터를 추가한다.

sampleData = [("p", 1), ("q", 2), ("9", 3), ("r", 4), ("8", 5)]
sc.parallelize(sampleData).sortBy(lambda x: x[0]).collect()
[('9', 3), ('8', 5), ('p', 1), ('q', 2), ('r', 4)]
sc.parallelize(sampleData).sortBy(lambda x: x[1]).collect()
[('p', 1), ('q', 2), ('9', 3), ('r', 4), ('8', 5)]
sc.parallelize(sampleData).sortBy(lambda x: x[1], ascending=False).collect()
[('8', 5), ('r', 4), ('9', 3), ('q', 2), ('p', 1)]

(8) 변환 연산 glom

동일한 파티션에 속한 요소들을 하나의 리스트로 결합한다. 생성된 리스트는 새 RDD의 단일 요소가 된다. glom 변환을 통해 원본 RDD의 파티션 상태를 확인할 수 있다.

dataset = [1, 2, 3, 4, 5, 6, 7, 8, 9]
rdd = sc.parallelize(dataset, 4)
rdd = rdd.glom()
rdd.collect()
[[1, 2], [3, 4], [5, 6], [7, 8, 9]]