AWS EC2部署的Apache NiFi 1.23.2日志整合压缩上传S3实现咨询

针对你通过HTTP发送日志到EC2上的NiFi，需要实现按100MB滚动生成文件、压缩后传S3的需求，具体步骤如下：

1. 配置HTTP日志接收 #

使用ListenHTTP处理器：

• 监听端口设为8080（和测试代码对应），上下文路径设为/logs
• 允许的HTTP方法勾选POST
• 确保EC2安全组开放该端口，允许日志发送端的IP访问

该处理器会将每一行POST过来的日志生成一个独立的FlowFile。

2. 按大小聚合日志文件 #

核心使用MergeContent处理器实现滚动文件：

• 合并策略：选择Bin Packing (By Size)
• 最大文件大小：设置为100 MB（或直接写104857600字节）
• 文件名格式：用log_${UUID()}.txt或log_${now():format("yyyyMMddHHmmssSSS")}.txt保证唯一性
• 可选配置：设置Idle Duration（比如10分钟），避免长时间无新日志时文件一直不输出

当聚合的文件大小达到100MB时，该处理器会自动输出完整的日志文件，并开始新的聚合桶，实现循环写入。

3. 压缩日志文件 #

添加CompressContent处理器：

• 压缩格式选择GZIP（S3存储推荐，节省空间）
• 输出文件名自动添加.gz后缀

4. 上传到AWS S3 #

使用PutS3Object处理器：

• 权限配置：给NiFi所在的EC2实例绑定IAM角色，授予目标S3桶的写入权限（无需硬编码密钥）
• 桶名：填写你的目标S3桶名称
• S3密钥前缀：可设置为logs/${now():format("yyyy/MM/dd")}/，按日期归档文件
• 文件名：保留压缩后的文件名即可

流程连接顺序 #

ListenHTTP → MergeContent → CompressContent → PutS3Object

Python测试代码（中文注释版） #

import requests
import os

# NiFi的HTTP接收地址，替换为你的EC2公网/内网IP
nifi_url = 'http://你的EC2_IP:8080/logs'

# 本地日志文件所在目录
directory_path = '/Users/arz/Desktop/logs/'

# 获取目录下所有文件
file_list = os.listdir(directory_path)

for file_name in file_list:
    file_path = os.path.join(directory_path, file_name)
    # 跳过子目录，只处理文件
    if os.path.isfile(file_path):
        with open(file_path, 'r') as file:
            # 逐行读取日志并发送到NiFi
            for line in file:
                line = line.strip()  # 去除换行符和首尾空白
                response = requests.post(nifi_url, data=line)
                if response.status_code == 200:
                    print(f"成功发送文件[{file_name}]中的行: {line}")
                else:
                    print(f"发送文件[{file_name}]中的行失败: {line}")
                    print(f"响应状态码: {response.status_code}")
                print(f"响应内容: {response.text}")
    else:
        print(f"跳过目录: {file_name}")

注意事项 #

• EC2权限：确保EC2实例的IAM角色拥有S3桶的PutObject权限，避免上传失败
• NiFi性能：如果日志量较大，可调整MergeContent的队列大小，避免阻塞
• 容错处理：可添加PutFile处理器作为备份，当S3上传失败时将文件暂存到EC2本地，后续重试

内容的提问来源于stack exchange，提问作者Erwin