使用Python从CSV文件匹配JSON日志条目并提取相关信息-Python教程-PHP中文网

使用Python从CSV文件匹配JSON日志条目并提取相关信息

DDD

发布： 2025-08-30 13:43:15

原创

370人浏览过

使用python从csv文件匹配json日志条目并提取相关信息

本文详细介绍了如何利用Python处理CSV和JSON两种不同格式的数据，实现基于CSV中IP地址和时间戳等关键信息，从JSON日志文件中筛选并提取匹配日志条目的需求。教程涵盖了数据读取、匹配逻辑构建、示例代码及性能优化等关键环节，旨在帮助读者高效地进行异构数据关联与分析。

在日常的数据处理工作中，我们经常会遇到需要从不同格式的数据源中关联信息的需求。例如，有一个包含用户行为或系统事件关键信息的CSV文件，而日志系统则以JSON格式记录了详细的运行日志。此时，我们可能需要根据CSV文件中的特定标识（如IP地址和时间戳）来筛选出JSON日志中对应的条目。本教程将指导您如何使用Python高效地完成这一任务。

1. 理解问题与挑战

我们的目标是：

读取一个CSV文件，其中包含 clientip 和 timestamp 等字段。
读取一个JSON日志文件，其中每行是一个JSON对象，包含日志信息。
如果JSON日志条目中的某些内容（例如，日志信息中包含的IP地址和日志时间）与CSV文件中的 clientip 和 timestamp 匹配，则将该JSON日志条目提取出来并保存到新文件。

挑战在于：

CSV和JSON的数据结构不一致，字段名称和组织方式可能不同。
JSON日志中的IP地址可能不是一个独立的字段，而是嵌入在某个字符串（如 info 字段）中。
需要高效地遍历和比较大量数据。

传统的Shell命令（如 join 或 awk）在处理结构化数据时非常强大，但对于这种需要解析复杂JSON结构并在字符串中进行模式匹配的场景，Python提供了更灵活和强大的解决方案。

立即学习“Python免费学习笔记（深入）”；

2. 核心工具：Python的csv和json模块

Python标准库中的 csv 和 json 模块是处理这两种数据格式的利器。

csv 模块：用于读写CSV（逗号分隔值）文件。csv.DictReader 特别方便，它将每一行读取为字典，其中键是CSV文件的标题行。
json 模块：用于编码和解码JSON（JavaScript对象表示法）数据。json.loads() 可以将JSON字符串解析为Python对象（如字典或列表），json.dumps() 则可以将Python对象转换为JSON字符串。

3. 数据准备与示例

为了演示，我们先模拟CSV和JSON数据。在实际应用中，您将直接从文件中读取这些数据。

易笔AI论文

专业AI论文生成，免费生成论文大纲，在线生成选题/综述/开题报告等论文模板

103

查看详情

示例 CSV 数据 (data.csv)

"clientip","destip","dest_hostname","timestamp"
"127.0.0.1","0.0.0.0","randomhost","2023-09-09T04:18:22.542Z"
"192.168.1.10","0.0.0.0","anotherhost","2023-09-09T05:00:00.000Z"

登录后复制

示例 JSON 日志数据 (logs.json)

[
    {"log": "09-Sept-2023", "rate-limit": "somethingelse?", "info": "client @xyz 127.0.0.1", "stream":"stderr", "time": "2023-09-09T04:18:22.542Z"},
    {"log": "09-Sept-2023", "rate-limit": "another_event", "info": "client @abc 192.168.1.10", "stream":"stdout", "time": "2023-09-09T05:00:00.000Z"},
    {"log": "10-Sept-2023", "rate-limit": "unrelated", "info": "client @def 10.0.0.1", "stream":"stderr", "time": "2023-09-10T06:00:00.000Z"}
]

登录后复制

注意：上述JSON数据是经过修正的，以确保其是有效的JSON格式，并且 time 字段被正确地表示为字符串。原始问题中提供的JSON示例可能存在格式问题，在实际处理前需要确保JSON数据的有效性。

4. 实现匹配逻辑

我们将通过以下步骤实现匹配：

读取CSV文件，逐行获取 clientip 和 timestamp。
读取JSON日志文件，解析所有日志条目。
对于CSV中的每一行，遍历JSON日志条目，检查 timestamp 是否精确匹配，并检查 clientip 是否作为子字符串存在于JSON日志条目的 info 字段中。
将所有匹配的JSON日志条目写入一个新的输出文件。

以下是实现此逻辑的Python代码：

import io
import csv
import json
import re # 导入re模块，用于更灵活的字符串匹配，尽管本例中'in'操作符已足够

def extract_matched_json_logs(csv_filepath, json_filepath, output_filepath):
    """
    根据CSV文件中的IP和时间戳，从JSON日志文件中提取匹配的日志条目。

    Args:
        csv_filepath (str): CSV文件的路径。
        json_filepath (str): JSON日志文件的路径。
        output_filepath (str): 匹配结果输出文件的路径。
    """
    matched_entries_count = 0

    try:
        # 1. 读取CSV数据
        # 使用'with'语句确保文件正确关闭
        with open(csv_filepath, 'r', encoding='utf-8') as csv_file:
            csv_reader = csv.DictReader(csv_file)
            csv_data_list = list(csv_reader) # 将CSV数据加载到内存中，方便多次遍历

        # 2. 读取JSON日志数据
        with open(json_filepath, 'r', encoding='utf-8') as json_file:
            json_data = json.load(json_file) # 加载整个JSON文件内容

        # 3. 遍历CSV数据并与JSON日志进行匹配
        with open(output_filepath, 'w', encoding='utf-8') as outfile:
            for csv_row in csv_data_list:
                csv_ip = csv_row.get('clientip')
                csv_timestamp = csv_row.get('timestamp')

                # 确保关键字段存在
                if not csv_ip or not csv_timestamp:
                    print(f"警告: CSV行缺少'clientip'或'timestamp'字段，跳过: {csv_row}")
                    continue

                for json_entry in json_data:
                    json_time = json_entry.get('time')
                    json_info = json_entry.get('info', '') # 使用.get()并提供默认值，防止KeyError

                    # 匹配条件：
                    # 1. 时间戳精确匹配
                    # 2. CSV中的IP地址作为子字符串存在于JSON日志的'info'字段中
                    if json_time == csv_timestamp and csv_ip in json_info:
                        # 找到匹配，将JSON条目写入输出文件
                        outfile.write(json.dumps(json_entry, ensure_ascii=False) + '\n')
                        matched_entries_count += 1
                        # 如果一个CSV行只应匹配一个JSON条目，可以在这里添加break
                        # break

        print(f"匹配完成！共找到 {matched_entries_count} 条匹配的JSON日志条目，已保存到 '{output_filepath}'。")

    except FileNotFoundError:
        print(f"错误: 文件未找到。请检查路径: {csv_filepath} 或 {json_filepath}")
    except json.JSONDecodeError:
        print(f"错误: JSON文件格式无效。请检查文件: {json_filepath}")
    except Exception as e:
        print(f"发生未知错误: {e}")

# --- 运行示例 ---
# 创建模拟文件
with open('data.csv', 'w', encoding='utf-8') as f:
    f.write('"clientip","destip","dest_hostname","timestamp"\n')
    f.write('"127.0.0.1","0.0.0.0","randomhost","2023-09-09T04:18:22.542Z"\n')
    f.write('"192.168.1.10","0.0.0.0","anotherhost","2023-09-09T05:00:00.000Z"')

with open('logs.json', 'w', encoding='utf-8') as f:
    f.write('''
[
    {"log": "09-Sept-2023", "rate-limit": "somethingelse?", "info": "client @xyz 127.0.0.1", "stream":"stderr", "time": "2023-09-09T04:18:22.542Z"},
    {"log": "09-Sept-2023", "rate-limit": "another_event", "info": "client @abc 192.168.1.10", "stream":"stdout", "time": "2023-09-09T05:00:00.000Z"},
    {"log": "10-Sept-2023", "rate-limit": "unrelated", "info": "client @def 10.0.0.1", "stream":"stderr", "time": "2023-09-10T06:00:00.000Z"}
]
''')

# 调用函数进行匹配
extract_matched_json_logs('data.csv', 'logs.json', 'matched_json_logs.txt')

登录后复制

5. 代码解析与注意事项

文件读取：
- csv.DictReader 将CSV文件的每一行转换为字典，方便通过列名访问数据。
- json.load() 用于读取整个JSON文件并解析为Python对象。如果JSON文件非常大，一行一个JSON对象（JSON Lines格式），则需要逐行读取并使用 json.loads() 解析每行。
错误处理：使用 try-except 块来捕获 FileNotFoundError 和 json.JSONDecodeError 等常见错误，提高程序的健壮性。
字典的 get() 方法：在访问字典字段时，使用 json_entry.get('time') 而不是 json_entry['time'] 是一个好习惯。get() 方法允许您指定一个默认值（例如 '' 或 None），以防止当键不存在时引发 KeyError。
IP地址匹配：由于JSON日志中的IP地址可能嵌入在字符串中，我们使用了 csv_ip in json_info 进行子字符串匹配。如果需要更精确的IP地址提取（例如，确保匹配的是完整的IP地址而不是部分数字），可以使用正则表达式 re.search(r'\b' + re.escape(csv_ip) + r'\b', json_info)。
输出文件：匹配到的JSON条目被写入 output_filepath 指定的文件。json.dumps(json_entry, ensure_ascii=False) 将Python字典转换回JSON字符串，ensure_ascii=False 确保中文字符不会被转义。\n 用于在每个JSON条目后添加换行符，使输出文件易于阅读。
性能考虑：
- 对于小型文件，将CSV和JSON数据一次性加载到内存中（如 csv_data_list = list(csv_reader) 和 json_data = json.load(json_file)）是可行的。
- 对于非常大的文件，一次性加载可能会导致内存不足。此时，可以考虑使用生成器（generator）逐行读取文件，或者将其中一个文件（通常是较小的那个）加载到内存中，然后流式处理另一个文件。例如，如果JSON日志文件非常大，您可以先将CSV数据构建成一个查找结构（如字典），然后逐行读取JSON日志并进行查找。