使用 pandas 链接数据转换方法的设计模式

Question

我每月收到一个包含一些列的 csv 文件。无论我收到什么列，如果可能的话，我应该输出一个包含 C1、C2、C3、... C29、C30 列的 csv + 一个包含我所采取的步骤的日志文件。

我知道，我的数据转换顺序应该是 t1, t2, t3, t4, t5。

t1 generates columns C8, C9, C12, C22 using C1, C2, C3, C4
t2 generates columns C10, C11, C17 using C3, C6, C7, C8
t3 generates columns C13, C14, C15, C16 using C5, C8, C10, C11, C22
t4 generates columns C18, C19, C20, C21, C23, C24, C25 using C13, C15
t5 generates columns C26, C27, C28, C29, C30 using C5, C19, C20, C21

我无法控制输入数据中的列。

如果我的输入数据有 C1、C2、C3、C4、C5、C6、C7 列，我可以生成所有 C1 ... C30 列。

如果我的输入数据有 C1、C2、C3、C4、C5、C6、C7、C8、C10、C11、C17 列，我可以生成所有 C1 ... C30 列，但我应该跳过 t2，因为它没必要

如果我的输入数据有 C1、C2、C3、C4、C6、C7，我只能做 t1、t2、t3、t4。我无法运行 t5，因此我应该只使用 NaN 值创建 C26、C27、C28、C29、C30 列，并且我应该在日志中添加“无法执行 t5 转换，因为缺少 C5。C26、C27、C28、C29、C30 是用 NaN 值填充”

我的 t1, t2, t3, t4, t5 已经创建好了，但是我不知道如何以优雅的方式组织代码以使代码重复最少。

我必须在很短的时间内开发我的代码。因此，我所有的 t1、t2、t3、t4、t5 方法看起来都像

def ti(df):
    output_cols = get_output_cols()
    if output_cols_already_exist(df, output_cols):
        return df, "{} skipped, the output cols {} already exist".format(inspect.stack()[0][3], output_cols)
    else:
        input_cols = get_required_input_cols()
        missing_cols = get_missing_cols(df, input_cols):
        if missing_cols == []:
            // do stuff
            log = "Performed {} transformation. Created {} columns".format(inspect.stack()[0][3], input_cols)
        else:
            for col in input_cols:
                df[col] = np.NaN
            log = "Cannot perform {} transformation because {} columns are missing. {} are filled with NaN values".format(inspect.stack()[0][3], missing_cols, output_cols)

另外，我使用这些函数的方式如下：

text = ""
df = pd.read_csv(input_path)
df, log_text = t1(df)
text = text + log_text + "\n"
df, log_text = t2(df)
text = text + log_text + "\n"
df, log_text = t3(df)
text = text + log_text + "\n"
df, log_text = t4(df)
text = text + log_text + "\n"
df, log_text = t5(df)
text = text + log_text + "\n"
df.to_csv("output_data.csv", index = False)
logging.info(text)

如您所见，我的代码丑陋且重复。现在我有时间重构它，但我不知道最好的方法是什么。我还希望我的代码是可扩展的，因为我也在考虑添加一个 t6 转换。你能帮我给出一些我可以遵循的方向/设计模式吗？ （除了 pandas，我也可以使用其他 python 库）

Answer 1

由于在 Python 中，函数是 first-class objects，因此您可以重构代码以便通过提取似乎可以区分它们的内容（do stuff 部分）来概括您的 t[i] 函数，使其成为辅助函数并处理它作为一个参数。

您还可以通过在列表上迭代来避免在调用函数（t1、t2 等或此后重构的帮助器版本）时重复。

最后，使用f-strings 有助于使您的代码更具可读性。

类似这样的：

# t function takes a dataframe and a function as parameters
def t(df, do_stuff_func):
    output_cols = get_output_cols()
    if output_cols_already_exist(df, output_cols):
        return (
            df,
            (
                f"{inspect.stack()[0][3]} skipped, "
                f"the output cols {output_cols} already exist",
            ),
        )
    else:
        input_cols = get_required_input_cols()
        missing_cols = get_missing_cols(df, input_cols)
        if missing_cols == []:
            # Call the helper function
            do_stuff_func()
            log = (
                f"Performed {inspect.stack()[0][3]} transformation."
                f"Created {input_cols} columns"
            )
        else:
            for col in input_cols:
                df[col] = np.NaN
            log = (
                f"Cannot perform {inspect.stack()[0][3]} transformation"
                f"because {missing_cols} columns are missing. "
                f"{output_cols} are filled with NaN values"
            )

# Define the five new 'do_stuff' functions
def do_stuff1():
    pass
...
def do_stuff5():
    pass

# Store the functions
do_stuff_funcs = [do_stuff1, do_stuff2, do_stuff3, do_stuff4, do_stuff5]

# Call t function in combination with df and do_stuff_funcs helpers
for do_stuff_func in do_stuff_funcs:
    df, log_text = t(df, do_stuff_func)
    text = text + log_text + "\n"

# Save the results
df.to_csv("output_data.csv", index = False)
logging.info(text)