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

Question

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

这是棘手的部分：处理是在 Cloud Dataflow 上完成的，这意味着我必须使用 Apache 光束变换来实现这一点。
我尝试了多种方法，例如 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 输入包含；

名称 custom1 custom2
arun undefined Nan
dany losangels 临时

预期的 csv；
名称 custom1 custom2
阿伦洛杉矶空
dany losangels 临时的

Answer 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")