Python映射两个csv文件

Question

我有一个配置文件（csv）：

Column name;Function;Args
Region;function1;arg1
Country;function2;arg1, arg2
email;function3;arg1
...

我想将我的配置文件中的特定功能应用到我的 csv 文件中的特定列（fileIn 大文件 > 1GB) 使用 dask、pandas 或标准 csv：

Region;Country;name
Europe;Slovakia;Mark
Asia;china;Steeve
...

有没有一种干净的方式来迭代配置文件？

df = pd.read_csv(fileIn, sep=';', low_memory=True, chunksize=1000000, error_bad_lines=False)

for chunk in df
    chunk['Region'] = chunk['Region'].apply(lambda x: MyClass.function1(args1))
    chunk['Country'] = chunk['Country'].apply(lambda x: MyClass.function2(arg1, arg2))
    chunk['email'] = chunk['email'].apply(lambda x: MyClass.function3(arg1))
    
    chunk['Region'].to_csv(fileOut, index=False, header=True, sep=';')
    ...

这是我在配置文件中调用的函数之一的示例：

def function1(value, replaceWith):
    text = re.sub(r'[^ ]', replaceWith, value)
    return text

Answer 1

您可以设置一个带有函数的字典，并在每次迭代时将其应用于块数据帧。

这里有一些代码，解释请看cmets：

# set up functions, for example
# - f1 to uppercase
# - f2 to lowercase
# - f3 to reverse string
def f1(x):
    return x.upper()

def f2(x):
    return x.lower()

def f3(x):
    return x[::-1]

# set up dict mapping function name in config to a function
fns = {
    'function1': f1,
    'function2': f2,
    'function3': f3,
}

# read config and set dict mapping column to a function
# here using the following `config.csv`:
#   Column name;Function
#   Region;function1
#   Country;function2
#   name;function3
df_config = pd.read_csv('config.csv', sep=';')
col_fns = df_config.set_index('Column name')['Function'].map(fns).to_dict()

# read and process csv in chunks
fileIn = 'file.csv'
fileOut = 'out.csv'
chunkSize = 1

df = pd.read_csv(
    fileIn, sep=';', low_memory=True, chunksize=chunkSize, error_bad_lines=False)

for i, chunk in enumerate(df):
    chunk_processed = chunk.apply(col_fns) # apply functions
    chunk_processed.to_csv(
        fileOut, index=False, header=(i==0), sep=';', mode='w' if i==0 else 'a')

# read the first 100 lines of the processed csv to test
pd.read_csv(fileOut, sep=';', nrows=100)

输出：

   Region   Country    name
0  EUROPE  slovakia    kraM
1    ASIA     china  eveetS

附：你需要把chunkSize改成合理的，当然这里只用1来测试，因为inFile只有2行，我们应该多块测试

---

更新：如果你需要应用的函数有一些参数，你可以创建包装函数：

# set up functions
def f(x, replaceWith):
    return re.sub(r'[^ ]', replaceWith, x)

# set up dict mapping function name in config to a function
fns = {
    'function1': f,
    'function2': f,
    'function3': f,
}

# read config and set dict mapping column to a function
# here using the following `config.csv`:
#   Column name;Function;args
#   Region;function1;A
#   Country;function2;B
#   name;function3;C
df_config = pd.read_csv('config.csv', sep=';')
col_fns = {r['Column name']: lambda x: fns[r['Function']](x, r['args'])
           for _, r in df_config.iterrows()}

...

# making it output both raw and processed values:
for i, chunk in enumerate(df):
    chunk_processed = chunk.apply(col_fns) # apply functions
    chunk_out = pd.concat(
        [chunk, chunk_processed.add_suffix('_processed')], axis=1)
    chunk_out.to_csv(
        fileOut, index=False, header=(i==0), sep=';', mode='w' if i==0 else 'a')

输出：

   Region   Country    name Region_processed Country_processed name_processed
0  Europe  Slovakia    Mark           AAAAAA          BBBBBBBB           CCCC
1    Asia     china  Steeve             AAAA             BBBBB         CCCCCC