如何计算给定 csv/psv 混合文本文件中的不同字段？

Posted 2023-04-15

【中文标题】如何计算给定 csv/psv 混合文本文件中的不同字段？

【英文标题】：How do I count the distinct fields from a given csv/psv hybrid text file?

【发布时间】：2020-01-23 22:46:59

【问题描述】：

我相信 Python 是最好的选择，但我可能是错的。

以下是来自 Linux 中文本格式数据源的示例：

TUI,39832020:09:01,10.56| TUI,39832020:10:53,11.23| TUI,39832020:15:40,23.20
DIAN,39832020:09:04,11.56| TUI,39832020:11:45,11.23| DIAN,39832020:12:30,23.20| SLD,39832020:11:45,11.22

大小未知，假设有一百万行。

每行包含三个或更多由| 分隔的集合，每个集合具有由, 分隔的字段。

每组中的第一个字段是产品 ID。例如，在上面的示例中，TUI、DIAN 和 SLD 是产品 ID。

我需要了解我存档的产品类型。例如，第一行包含 1：TUI，第二行包含 3：DIAN、TUI 和 SLD。

总的来说，在这两条线上，我们可以看到三个独特的产品。

谁能帮忙？

非常感谢。感谢任何启发。

更新

我更喜欢基于 Python 和 Spark 的解决方案，即 pySpark。

我也在寻找如下统计数据：

每种产品的总金额； 给定时间的所有记录（每组中的第二个字段，如39832020:09:01）； 每种产品的最低和最高价格。

更新 2

谢谢大家的代码，我真的很感激。我想知道是否有人可以将数据写入 RDD 和/或数据帧。我知道在 SparkSQL 中获取这些统计信息非常简单。

非常感谢。

非常感谢。

【问题讨论】：

我想我明白了你所说的一切，并为每个任务用 sn-ps 更新了我的答案。检查一下，如果您需要其他东西，请告诉我。

39832020:09:01 中的39832020 是什么？第二部分我可以推断是时间，但我无法弄清楚前半部分。

你已经尝试了什么？请检查"How to create a Minimal, Complete, and Verifiable example" 和"How to ask"。按照这些文章的提示，您将获得更好的结果。尽管我为您提供了替代方案，但您需要向我们展示您自己做了什么，然后我们可以帮助您走得更远。我们无法为您编写解决方案。 SO 不适合那种事情。

【参考方案1】：

类似于 Accdias 的答案：使用字典，逐行读取文件，将数据按 | 拆分，然后按 , 拆分，然后将字典中的计数加起来。

myFile="lines_to_read.txt"
productCounts = dict()

with open(myFile, 'r') as linesToRead:
    for thisLine in linesToRead:
        for myItem in thisLine.split("|"):
            productCode=myItem.split(",")
            productCode=productCode[0].strip()
            if productCode in productCounts:
                productCounts[productCode]+=1
            else:
                productCounts[productCode]=1

print(productCounts)

**** 更新 **** Dataframe 与 Pandas 一起使用，以便我们可以查询数据后记的统计信息：

import pandas as pd

myFile="lines_to_read.txt"
myData = pd.DataFrame (columns=['prodID', 'timeStamp', 'prodPrice'])

with open(myFile, 'r') as linesToRead:
    for thisLine in linesToRead:
        for myItem in thisLine.split("|"):
            thisItem=myItem.strip('\n, " "').split(",")
            myData = myData.append('prodID':thisItem[0],'timeStamp':thisItem[1],'prodPrice':thisItem[2], ignore_index=True)

print(myData)   # Full Table
print(myData.groupby('prodID').agg('prodID':'count'))  # Total of prodID's
print(myData.loc[myData['timeStamp'] == '39832020:11:45']) # all lines where time = 39832020:11:45
print(myData.groupby('prodID').agg('prodPrice':['min', 'max'])) # min/max prices

【讨论】：

谢谢Brian，是否可以提供RDD/Dataframe 解决方案？

@mdivk 挑战已接受，答案已更新。请注意，此代码非常精简，几乎没有错误处理，并且基于一个很小的数据集。

谢谢布赖恩，我要求 DataFrame 解决方案的原因是我喜欢通过将 df 注册为 TempTable 来进行临时查询，我知道 pandas 数据框中有 API 可以做到这一点，但是，我个人认为 TempTable 是快速完成这项工作的最简单方法