pySpark5- filter, min/max

temperature.csv

map 함수의 특징:

map 함수는 RDD(Resilient Distributed Dataset)나 DataFrame의 각 요소에 주어진 함수를 적용하여 새로운 RDD 또는 DataFrame을 반환합니다.
map 함수는 변환된 결과를 바로 반환해야 하므로, yield를 사용하면 에러가 발생합니다.
yield는 제너레이터를 생성하여 값을 순차적으로 반환하는 방식이기 때문입니다.

yield 대신 return 사용: get_data 함수 내에서 yield 대신 return을 사용하여 변환된 결과를 직접 반환합니다.
flatMap 사용: 여러 개의 값을 반환해야 하는 경우 flatMap 함수를 사용합니다. flatMap은 각 요소에 함수를 적용한 후, 결과를 다시 평평하게 펼쳐 하나의 RDD 또는 DataFrame으로 반환합니다. 이때 get_data 함수는 yield를 사용하여 여러 값을 생성할 수 있습니다.

import pyspark

sc = pyspark.SparkContext.getOrCreate();
test_file = "file:///home/jovyan/work/sample/temperature.csv"

def get_data(line, header):
    if line != header:
        col = line.split(',')
        city = col[6].strip("\"")
        avg_temp_fahr = col[4]
        yield (city, avg_temp_fahr)

lines = sc.textFile(test_file)

header = lines.first()

parsed_line = lines.flatMap(lambda line: get_data(line, header))


filtered_line = parsed_line.filter(lambda x: "NA" not in x[1])

min_temp = filtered_line.reduceByKey(lambda x, y: min(float(x), float(y)))
final_list = min_temp.collect();
for city, temperature in final_list:
    print(f"{city}: {temperature}")