一、前言

尽管自动化安全工具以及第一层和第二层安全运营中心 (SOC) 分析师应该能够处理大约 80% 的威胁，但我们仍需对余下的20%保持高度的警惕。这部分未被充分覆盖的威胁中，往往隐藏着更为复杂且可能带来重大损失的潜在风险。

网络威胁狩猎为企业安全带来人为元素，补充了自动化系统的不足。凭借丰富的人为经验能够在威胁可能导致严重问题之前发现、记录、监控并消除威胁。

二、简单的ssh异常登陆检测

假设我们要构建一个最常见的异常检测场景：从海量的SSH连接日志中筛选出异常连接。

那么，如何界定SSH连接的异常性呢？我们可以从多个维度入手，比如时间维度（例如深夜的非正常工作时间段）、机器属性（如机器归属人、归属部门）以及历史记录等。这里，我们选择通过历史记录进行筛选，来做一个简单的实践。具体来说，我们可以统计今天的SSH连接记录，找出那些在过去一个月中从未出现过的连接，作为初步的异常检测。

虽然这个思路看似简单，但在实际操作中，却需要使用大数据处理组件来进行。毕竟，一个月的SSH连接数据量庞大，而且每条SSH日志除了包含IP等基本信息外，还包含其他丰富的附加信息，整体数据量大概10G。

我使用mac M3 36G机器上对10GB的ip.txt（模拟生成的）进行简单去重。

import time
from collections import Counter


with open("ip.txt",encoding="gbk") as f:
    content = f.read()
    start_time = time.time()


    word_list = content.split(" ")
    word_counts = Counter(word_list)
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"Execution time: {execution_time} seconds")
    print(word_counts)

使用python进行处理，内存直接爆炸了，就更不用说处理时间方面了（脚本都没跑完）

那如何解决呢？

三、站在巨人的肩膀上spark

剖析一下为何此次操作会以失败告终：原因在于，我试图一次性将10GB的庞大数据文件全部加载到内存中，随后使用Python进行split、Counter等处理操作，这无疑导致内存使用量急剧飙升，最终因内存耗尽而引发程序崩溃。

为了优化这一流程，我们应当采取分段读取的策略，即每次仅读取文件的一部分数据，并对其进行相应的计算处理。同时，我们还可以利用多线程技术来充分压榨CPU的性能，从而提升运算效率（当然，这一切都需要在严格的内存管理之下进行）。

事实上，上述这些繁琐的步骤已经有人为我们提前做好了：Apache Spark 是一种用于大数据工作负载的分布式开源处理系统。它使用内存中缓存和优化的查询执行方式，可针对任何规模的数据进行快速分析查询。

比如我们生成的一个亿ssh连接日志

import random


with open("temp.csv","w") as f:
    for _i in range(1,10000000):
        host_ip = "10.0.{}.{}".format(str(random.randint(0, 255)),str(random.randint(0, 255)))
        dst_ip = "10.0.{}.{}".format(str(random.randint(0, 255)), str(random.randint(0, 255)))
        # print(f"log_service,1734770440,aegis-log-