MRjob处理CSV含逗号字段的拆分问题及解决方案（Python）

我用Python的MRJob编写MapReduce代码处理CSV数据集，数据集表头为Year、Length、Title、Genre、Actor、Actress、Director、Popularity、Awards、Image，部分数据行示例：

• 行1：1990,111,Tie Me Up! Tie Me Down!,Comedy,"Banderas, Antonio","Abril, Victoria","Almodóvar, Pedro",68,No,"NicholasCage.png,,"
• 行2：1991,113,High Heels,Comedy,"Bosé, Miguel","Abril, Victoria","Almodóvar, Pedro",68,No,"NicholasCage.png,,"

代码目标是统计1970年后上映、时长超过75分钟的各类型电影数量，但最初用逗号拆分数据时，部分字段（如演员、导演）包含带逗号的内容，导致拆分后列错位。

原错误代码 #

class MovieGenreCount(MRJob):

    def mapper(self, _, line):
        movie = line.split(',')
        year = int(movie[0])
        length = int(movie[1])
        genre = movie[2]
        if year > 1970 and length > 75:
            yield genre, 1
    
    def combiner(self, genre, counts):
        yield genre, sum(counts)
    
    def reducer(self, genre, counts):
        yield genre, sum(counts)
        
if __name__ == '__main__':
    MovieGenreCount.run()

原代码问题 #

直接使用line.split(',')拆分CSV行，无法识别被双引号包裹的含逗号字段，会将这类字段拆分为多个元素，导致后续列的索引全部偏移，最终统计结果完全错误。

修正后的解决方案代码 #

import csv

class MovieGenreCount(MRJob):

    def mapper(self, _, line):
        row = list(csv.reader([line]))[0]
        year, length, title, genre, actor, actress, director, popularity, awards, image = row
        if year and length:
            year = int(year)
            length = int(length)
            if year > 1970 and length > 75:
                yield genre, 1
    
    def combiner(self, genre, counts):
        yield genre, sum(counts)
    
    def reducer(self, genre, counts):
        yield genre, sum(counts)
        
if __name__ == '__main__':
    MovieGenreCount.run()

解决方案说明 #

1. 引入Python标准库的csv模块，使用csv.reader解析单行数据，它能正确识别被双引号包裹的含逗号字段，确保每列数据的位置准确。
2. 增加year和length的非空判断，避免空值转换为整数时抛出异常，提升代码健壮性。

内容的提问来源于stack exchange，提问作者hadi khan