使用数据流清理CSV文件中的数据

问题描述 投票:0回答:1

我正在尝试从GCS读取一个CSV(带标题)文件,该文件大约有150列然后 1.设置特定列的列数据 2.使用所有列的Null值更新NaN 3.将csv文件(带标头)写入GCS

这是一个棘手的部分:处理是在Cloud Dataflow上完成的,这意味着我必须使用Apache beam变换来实现这一目标。 我尝试过多种方法,比如skipping_header_lines和使用架构

我的管道代码是:


def parse_method(self, line):    
    reader = csv.reader(line.split('\n'))
    for csv_row in reader:
        values = [x.decode('utf8') for x in csv_row]
        row = []
        for value in csv_row:
            if value == 'NaN':
                value = 'Null'
            row.append(value)
    return row

(p
    | 'Read_from_source' >> beam.io.ReadFromText('gs://{0}/test.csv'.format(BUCKET))
    | 'Split' >> beam.Map(lambda s: data_ingestion.parse_method(s))
    | 'Write_to_dest' >> beam.io.WriteToText(output_prefix,file_name_suffix='.csv', num_shards=1))

例如:如果我的csv输入包含;

name custom1 custom2 arun undefined南 dany losangels临时

预期的csv; name custom1 custom2 arun losangels Null dany losangels临时

python google-cloud-platform google-cloud-storage google-cloud-dataflow
1个回答
1
投票

使用以下内容生成您正在寻找的输出:

    lines = p | ReadFromText(file_pattern="gs://<my-bucket>/input_file.csv")

    def parse_method(line):
        import csv
        reader = csv.reader(line.split('\n'))
        for csv_row in reader:
            values = [x.decode('utf8') for x in csv_row]
            row = []
            for value in csv_row:
                if value == 'NaN':
                    value = 'Null'
                row.append(value)

        return ",".join(row)



    lines = lines | 'Split' >> beam.Map(parse_method)
    line = lines | 'Output to file' >> WriteToText(file_pattern="gs://<my-bucket>/output_file.csv")

现在根据标题编辑列我不确定是否有更简单的方法,但我会用以下方式使用pandas:

    lines = p | "ReadFromText" >> ReadFromText(file_pattern="gs://<my-bucket>/input_file.csv")

    def parse_method(line):
        import pandas as pd

        line = line.split(',')
        df = pd.DataFrame(data=[line],columns=['name','custom1','custom2'])
        df['custom2'] = df.custom2.apply(lambda x: 'None' if x == 'Nan' else x)
        values = list(df.loc[0].values)
        return ",".join(values)

    lines = lines | "Split" >> beam.Map(parse_method)
    line = lines | "Output to file" >> WriteToText(file_path_prefix="gs://<my-bucket>/output_file.csv")
© www.soinside.com 2019 - 2024. All rights reserved.