Comment utiliser le fonctionnement en temps réel Python3 pour traiter les fichiers journaux

1. Traitement simple de fichiers en temps réel (fichier unique)

Supposons que le chemin du journal que nous voulons lire en temps réel est : /data/mongodb/shard1/log/pg.csv

Ensuite, nous pouvons utiliser le script shell dans le fichier python Command tail -F pour la lecture et le fonctionnement en temps réel

Le code est le suivant :

import re
import codecs
import subprocess
 
def pg_data_to_elk():
    p = subprocess.Popen('tail -F /data/mongodb/shard1/log/pg.csv', shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,)    #起一个进程，执行shell命令
    while True:
        line = p.stdout.readline()   #实时获取行
        if line:                     #如果行存在的话
            xxxxxxxxxxxx
            your operation

Une brève explication du module de sous-processus :

le sous-processus vous permet de générer de nouveaux processus, de vous connecter à leur entrée /output/error pipes et obtenez leur code de retour (statut).

Introduction à subprocess.Popen

Cette classe est utilisée pour exécuter un sous-programme dans un nouveau processus.

Constructeur de subprocess.Popen

class subprocess.Popen(args, bufsize=-1, executable=None, stdin=None, stdout=None, stderr=None, 
    preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, universal_newlines=False,
    startup_info=None, creationflags=0, restore_signals=True, start_new_session=False, pass_fds=())

Description du paramètre :

• args : La commande shell à exécuter peut être une chaîne ou une séquence composée de chaque paramètre de la commande. Lorsque la valeur de ce paramètre est une chaîne, le processus d'interprétation de la commande dépend de la plateforme, il est donc généralement recommandé de transmettre le paramètre args sous forme de séquence.

• stdin, stdout, stderr : représentent respectivement les poignées d'entrée, de sortie et d'erreur standard du programme.

• shell : Ce paramètre est utilisé pour identifier s'il faut utiliser le shell comme programme à exécuter. Si la valeur du shell est True, il est recommandé de transmettre le paramètre args sous forme de chaîne plutôt que sous forme de séquence.

2. Traitement complexe de fichiers en temps réel (génération continue de nouveaux fichiers)

Si le journal génère un nouveau fichier journal lorsque certaines conditions sont remplies, par exemple, log1.csv a atteint 20 Mo, alors log2.csv le fera soit écrit, il y a probablement plus de 1 000 fichiers par jour, et de nouveaux sont constamment générés. Alors comment les obtenir en temps réel ?

L'idée est la suivante :

Ajoutez la détermination de la taille du fichier actuel dans la surveillance en temps réel (tail -F) Si la taille du fichier actuel est supérieure à 20 Mo, sortez de la surveillance en temps réel et obtenez-en une nouvelle. fichiers journaux. (C'est la même idée s'il existe d'autres conditions de jugement, mais remplacez le jugement de la taille actuelle du fichier par le jugement dont vous avez besoin)

Le code est le suivant :

import re
import os
import time
import codecs
import subprocess
from datetime import datetime
 
path = '/home/liao/python/csv'
time_now_day = datetime.now.strftime('%Y-%m-%d')
 
def get_file_size(new_file):
    fsize = os.path.getsize(new_file)
    fsize = fsize/float(1024*1024)
    return fsize
 
def get_the_new_file():
    files = os.listdir(path)
    files_list = list(filter(lambda x:x[-4:]=='.csv' and x[11:21]==time_now_day, files))
    files_list.sort(key=lambda fn:os.path.getmtime(path + '/' + fn) if not os.path.isdir(path + '/' + fn) else 0)
    new_file = os.path.join(path, files_list[-1])
    return new_file
 
def pg_data_to_elk():
    while True:
        new_file = get_the_new_file()
        p = subprocess.Popen('tail -F {0}'.format(new_file), shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,)    #起一个进程，执行shell命令
        while True:
            line = p.stdout.readline()   #实时获取行
            if line:                     #如果行存在的话
                if get_file_size(new_file) > 20:    #如果大于20M，则跳出循环
                    break
                xxxxxxxxxxxx
                your operation
        time.sleep(3)

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!