合并具有相同ID但不同序列号的行

Question

我有一个大型（15m 行）csv 文件，其中一些 ID 具有多个期限（STRT_DTE 和 EXP_DATE）。我还有一个名为 TRM_NBR 的列，可以取值 1、2、3 等。连续的 TRM_NBR 表示连续的术语周期（即，具有相同 ID 的三行和 TRM_NBR 1、2、3 是一个连续周期，而 1、2、 1 是两个句号）。

我需要减少 csv 文件，以便每个连续的期限（和 ID）有一行。请参见下面的示例。具体来说，我需要输出序列中first term的ID、STRT_DTE，以及序列last term的EXP_DTE。 （保留 TRM_NBR 是可选的。）

原始数据：

ID;TRM_NBR;STRT_DTE;EXP_DTE
000020000007;1;2017-08-22 00:00:00.000;2016-09-20 00:00:00.000
000020000009;1;2015-07-23 00:00:00.000;2015-03-24 00:00:00.000
000020000017;1;2014-10-02 00:00:00.000;2014-10-02 00:00:00.000
000020000063;1;2018-11-19 00:00:00.000;2018-11-19 00:00:00.000
000020000063;2;2020-11-19 00:00:00.000;2020-11-19 00:00:00.000
000020000356;1;2020-06-23 00:00:00.000;2020-06-23 00:00:00.000
000020000356;2;2021-05-20 00:00:00.000;2021-05-20 00:00:00.000
000020000356;3;2022-04-21 00:00:00.000;2021-10-21 00:00:00.000
000020000356;2;2014-07-22 00:00:00.000;2014-09-09 00:00:00.000
000020000356;3;2015-07-21 00:00:00.000;2015-07-21 00:00:00.000
000020000356;4;2016-07-12 00:00:00.000;2016-07-12 00:00:00.000
000020000356;5;2017-07-11 00:00:00.000;2017-07-11 00:00:00.000

期望的输出：

ID;TRM_NBR;STRT_DTE;EXP_DTE
000020000007;1;2017-08-22 00:00:00.000;2016-09-20 00:00:00.000
000020000009;1;2015-07-23 00:00:00.000;2015-03-24 00:00:00.000
000020000017;1;2014-10-02 00:00:00.000;2014-10-02 00:00:00.000
000020000063;1;2018-11-19 00:00:00.000;2020-11-19 00:00:00.000
000020000356;1;2020-06-23 00:00:00.000;2021-10-21 00:00:00.000
000020000356;2;2014-07-22 00:00:00.000;2017-07-11 00:00:00.000

连续的周期可以从 1 到无穷大，可能以 1 开头，也可能不以 1，但总是以 1 递增（如果有不止一行）。输出文件可以包含相同 ID（但序列不同）的多行。您可以假设文件按顺序正确排序。

我可以使用 Python、MySQL 或 Mac Terminal 工具来解决。

Answer 1

基于标准itertools 和第三方convtools 库的Python 替代方案：

import itertools

from convtools import conversion as c
from convtools.contrib.tables import Table

# read rows into iterable of dicts
rows = Table.from_csv(
    "input.csv", header=True, dialect=Table.csv_dialect(delimiter=";")
).into_iter_rows(dict)
# group into sequences of IDs
key_to_rows = itertools.groupby(rows, key=lambda r: r["ID"])

# prepare the converter, which takes groups of rows from the
# previous step and reduces each group to one row.
# This is where code generation happens, consider storing the
# resulting converter somewhere for further reuse.
converter = c.iter(
    c.item(1).pipe(
        c.aggregate(
            {
                "ID": c.ReduceFuncs.First(c.item("ID")),
                "LAST_TRM_NBR": c.ReduceFuncs.Last(c.item("TRM_NBR")),
                "STRT_DTE": c.ReduceFuncs.First(c.item("STRT_DTE")),
                "EXP_DTE": c.ReduceFuncs.Last(c.item("EXP_DTE")),
            }
        )
    )
).gen_converter()

# process iterable from itertools
processed_rows = converter(key_to_rows)

# output processed rows to "out.csv" (consider passing same dialect
# as above to .into_csv method
Table.from_rows(processed_rows).into_csv("out.csv")

输出是：

ID,LAST_TRM_NBR,STRT_DTE,EXP_DTE
000020000007,1,2017-08-22 00:00:00.000,2016-09-20 00:00:00.000
000020000009,1,2015-07-23 00:00:00.000,2015-03-24 00:00:00.000
000020000017,1,2014-10-02 00:00:00.000,2014-10-02 00:00:00.000
000020000063,2,2018-11-19 00:00:00.000,2020-11-19 00:00:00.000
000020000356,5,2020-06-23 00:00:00.000,2017-07-11 00:00:00.000

Answer 2

MySQL 的解决方案。

1. 为导入的源数据创建表：

CREATE TEMPORARY TABLE tmp (
    rowno INT AUTO_INCREMENT PRIMARY KEY,
    id CHAR(12),
    trm_nbr TINYINT,
    strt_dte DATETIME(3),
    exp_dte DATETIME(3)
) ENGINE = Memory;

1. 将源数据导入其中：

LOAD DATA INFILE 'X:/folder/filename.CSV'
INTO TABLE tmp
COLUMNS TERMINATED BY ';'
LINES TERMINATED BY '\r\n'
IGNORE 1 LINES
(id, trm_nbr, strt_dte, exp_dte);

1. 处理数据并将其保存到输出文件：

WITH
cte1 AS ( SELECT *,
                 LAG(id) OVER (ORDER BY rowno) lag_id, 
                 1 + LAG(trm_nbr) OVER (ORDER BY rowno) lag_trm_nbr
          FROM tmp ),
cte2 AS ( SELECT *,
                 SUM(CASE WHEN (id, trm_nbr) = (lag_id, lag_trm_nbr)
                          THEN 0
                          ELSE 1 
                          END) OVER (ORDER BY rowno) grp_no
          FROM cte1 )
SELECT 'ID', 'TRM_NBR', 'STRT_DTE', 'EXP_DTE'
UNION ALL
SELECT MAX(id) id, 
       ROW_NUMBER() OVER (PARTITION BY MAX(id) ORDER BY grp_no) trm_nbr, 
       MIN(strt_dte) strt_dte, 
       MAX(exp_dte) exp_dte
FROM cte2
GROUP BY grp_no
INTO OUTFILE 'X:/folder/new_filename.CSV'
COLUMNS TERMINATED BY ';'
LINES TERMINATED BY '\r\n';

DEMO

---

您当前的 MySQL 帐户必须具有 FILE 权限。

secure_file_priv 会话变量值必须是空字符串或某个drive:path（通过SELECT @@secure_file_priv; 查询检查）。如果不是空字符串，则查询中的X:/folder 必须等于该值（在 Windows 系统上，路径值中的反斜杠必须加倍或替换为直斜杠）。

如果secure_file_priv 会话变量值为NULL，则根本不能使用此方法。在这种情况下，请尝试要求服务器管理员设置此变量并允许与文件系统交互。

查询使用 Windows 样式 CSV 的设置，Unix 样式文件使用 LINES TERMINATED BY '\n'。

如果您的源文件太大（大于 ~100 MB 或大于 max_allowed_packet 会话变量值），则从表定义中删除 ENGINE = Memory。

您不需要删除临时表 - 当您关闭连接时它会自动删除。但是你可以明确地放弃它。