Liste du code Python utilisé dans l'analyse de Big Data

Environnement

• Créez un environnement virtuel au besoin.

Créer un environnement virtuel

$ python3 -m venv test

Activation de l'environnement virtuel

$ source test/bin/activate
(test)$

Désactiver l'environnement virtuel

(test)$ deactivate
$

Traitement des chaînes

Découpez une chaîne de caractères

• [Index de début: numéro de caractère de fin (notez qu'il ne s'agit pas d'un index)].

s = "2019-06-01"
print(f"{s[0:4]}-{s[5:7]}-{s[8:10]}")

échapper

Échapper aux crochets ondulés

--Echappez les crochets ondulés avec des crochets ondulés.

var = 'aiuto'
print( f"val is {{{var}}}" )

Manipulation d'annuaire

Créer un répertoire

import os

os.makidirs('tmp', exist_ok=True)

class

• À utiliser lorsque vous souhaitez séparer des propriétés, etc.

classsample
├── main.py
└── prop
    └── user_property.py

main.py

from prop.user_property import UserProperty

user_property = UserProperty({'first_name': 'Ichiro', 'family_name': 'tester'})
print(f'{user_property.FAMILY_NAME} {user_property.FIRST_NAME}')

prop/user_property.py

from typing import ClassVar, Dict, List, Any

class UserProperty:
    def __init__(self, kwargs: Dict[str, Any]):
        self.FIRST_NAME = kwargs['first_name']
        self.FAMILY_NAME = kwargs['family_name']

Résultat d'exécution

$ python main.py
Tester Ichiro

subprocess

• Vous pouvez exécuter des commandes shell à partir de Python avec un sous-processus.
• Au contraire, les données volumineuses ne peuvent pas être analysées sans exécuter des commandes shell.

Exécuter la commande shell

import subprocess

c = ['hadoop', 'fs', '-rm', '-r', '/tmp/test']
subprocess.run(c)

Exécution parallèle de commandes shell avec xargs

• Vous devez définir shell = True pour utiliser des tuyaux.
• Vous pouvez recevoir le processus avec subprocess.Popen () et attendre la fin du traitement avec wait ().
• Le traitement ultérieur ne sera pas exécuté tant que le processus ne sera pas terminé.
• L'exemple contient chaque fichier en tmp de Python et exécute test.sh avec 10 parallèles.
• À utiliser lorsque vous souhaitez raccourcir le temps de traitement en exécutant uniquement le traitement intermédiaire en parallèle.

c = 'ls tmp/* | xargs -L 1 -P 10 -t bash -c \'cat $0 | test.sh -'
p = subprocess.Popen(c, shell = True)
p.wait()

#Traitement ultérieur

Gestion de la sortie standard

• Ne passez pas stdout et stderr à pipe si vous n'en avez pas besoin.
• Vous pouvez recevoir une sortie standard avec python test.py &> log / test.log.

click

• Vous pouvez facilement implémenter des commandes qui peuvent être exécutées dans le terminal en cliquant sur.
• Implémentez la commande avec @ click.command (). --Plusieurs commandes peuvent être implémentées avec @ click.group () et add_command ().
• Vous pouvez ajouter des arguments de commande avec @ click.option ().

click
├── cli.py
└── command
    └── hello
        └── cli.py

click/cli.py

import click

from command.hello.cli import hello

@click.group()
def entry_point():
    print('click/cli.message py.')

entry_point.add_command(hello)

def init():
    entry_point(**{})

if __name__ == '__main__':
    init()

click/command/hello/cli.py

import click

@click.command('hello')
@click.option('--msg', '-m', 'msg', type=str, help='Saisissez le message que vous souhaitez afficher.')
def hello(**kwargs):
    print(f'Message entré:{kwargs["msg"]}')
    print('click/cmd/hello/cli.message py.')

$ python cli.py hello -m 'tester'
click/cli.message py.
Message entré: Test
click/cmd/hello/cli.message py.

pandas

• Utilisé lors du traitement des données.

tsv lire

• Vous pouvez spécifier un délimiteur avec un délimiteur.
• Vous pouvez définir des noms de colonnes avec des noms.
• Vous pouvez spécifier le type de données avec dtype. --Set low_memory = False lors de la gestion de fichiers volumineux.

import pandas as pd

df = pd.read_csv('user.tsv', delimiter='\t', header=None, names=['id', 'name'], dtype={'id': str, 'name': str}, low_memory=False)

sortie tsv

df.to_csv('test.tsv', sep='\t')

Sortie en spécifiant une colonne spécifique

• Utilisé lorsque vous souhaitez organiser les colonnes nécessaires à l'analyse, mais qui ne le sont pas.

columns = ['id', 'name']
df[colums].to_csv('test.tsv', sep='\t', index=False)

Sortie en réduisant le nombre de cas

• Utilisé pour l'échantillonnage.

df.sample(n=100).to_csv('test.tsv', sep='\t')

Suppression de ligne en double

df.drop_duplicates()

Utiliser des guillemets doubles dans la requête

df.query('row_name.str.contains("\\\"keyword\\\"")')

La gestion des erreurs

Élimination du script Python

• À utiliser lorsque vous souhaitez supprimer un script Python lorsqu'une erreur se produit.

import sys

sys.exit(1)

Vérification de l'existence du fichier

―― Utilisé pour voir si vous disposez des entrées nécessaires avant d'effectuer l'analyse des données.

import os

if os.path.exists():
    print('Le fichier existe. Effectue le traitement ultérieur.')
else:
    print('Le fichier n'existe pas. Le processus se termine.')
    sys.exit(1)

Enregistrement

--Utilisez la journalisation du module standard de Python.

• Un exemple de journalisation avec la configuration suivante est décrit.

test
├── module
│   └── sub.py
└── main.py

main.py

#Module fait maison
import module.sub as sub

from logging import CRITICAL, DEBUG, ERROR, INFO, WARNING
from logging import NullHandler, StreamHandler, basicConfig, getLogger, Formatter
from logging.handlers import TimedRotatingFileHandler

logger = getLogger(__name__)
logger.addHandler(NullHandler())
logger.setLevel(DEBUG)
sh = StreamHandler()

def init() -> None:
    basicConfig(
        handlers=[sh],
        format="[%(asctime)s] %(name)s %(levelname)s: %(message)s",
        datefmt="%y-%m-%d %H:%M:%S",
    )
    root_logger = getLogger()
    root_logger.setLevel(DEBUG)

    rfh = TimedRotatingFileHandler(
        "log/test.log",
        when="midnight",
        backupCount=30,
    )
    format_template = (
        f"PID:%(process)d [%(asctime)s] %(name)s %(levelname)s: %(message)s"
    )
    log_format = Formatter(fmt=format_template, datefmt="%y-%m-%d %H:%M:%S")
    rfh.setFormatter(log_format)
    root_logger.addHandler(rfh)

    logger.debug("Lancer l'exécution du script")

if __name__ == "__main__":
    init()
    #Appelez la fonction de votre propre module
    sub.hello()

module/sub.py

from logging import getLogger

logger = getLogger(__name__)

def hello():
    print('hello! this is sub module.')
    logger.debug('Sortie du sous-module')

$ python main.py
[20-06-25 14:20:56] __main__ DEBUG:Lancer l'exécution du script
hello! this is sub module.
[20-06-25 14:20:56] module.sub DEBUG:Sortie du sous-module

$ head log/test.log
PID:15171 [20-06-25 14:20:56] __main__ DEBUG:Lancer l'exécution du script
PID:15171 [20-06-25 14:20:56] module.sub DEBUG:Sortie du sous-module

Autre

Obtenez le nombre de fichiers

• Peut être obtenu en une ligne sans exécuter de commande shell.

cnt = str(sum(1 for line in open('test.tsv')))

Faire du fichier une chaîne d'une ligne

• Utilisé lorsque plusieurs mots-clés sont combinés en une seule ligne sous la condition OR.

main.py

import os

def load_file_as_one_line(file, sep):
    with open(file) as f:
        lines_one_str = ''
        # a\nb\nc\n -> a|b|c|d
        lines = f.readlines()
        for line in lines:
            w = line.rstrip(os.linesep)
            if(w != ''):
                lines_one_str += w + sep
        return lines_one_str[:-1]

print(load_file_as_one_line('data.txt', '|'))

$ cat data.txt
tester
test
texte
text
Goût
$ python main.py
tester|test|texte|text|Goût|taste

Générer dynamiquement un répertoire pour la partition de date

• Dans l'analyse des données, il existe de nombreuses situations dans lesquelles la valeur de n mois de données est lue et la valeur de n jours de données est lue, elle est donc utilisée dans de tels cas.

main.py

import datetime
from dateutil.relativedelta import relativedelta

def out_term(year, month, term, base_dir):
    d = datetime.date(year, month, 1)
    txt = ""

    for i in range(term):
        txt += base_dir + (d + relativedelta(months=i)).strftime("%Y/%m")
        if(i != term - 1) :
            txt += ","
    return txt
    
def out_reverse_term_by_day(d, reverse_term, base_dir):
    txt = ""
    
    d = d - relativedelta(days=reverse_term - 1)
    for i in range(reverse_term):
        txt += base_dir + (d + relativedelta(days=i)).strftime("%Y/%m/%d")
        if(i != reverse_term - 1) :
            txt += ","
    return txt

# 2019-Préparer un annuaire pendant 11 à 4 mois
print(out_term(2019, 11, 4, '/tmp/input/'))
# 2019-11-Préparez un annuaire qui remonte à 5 jours depuis 02
print(out_reverse_term_by_day(datetime.date(2019, 11, 2), 5, '/tmp/input/'))

Résultat d'exécution

$ python main.py
/tmp/input/2019/11,/tmp/input/2019/12,/tmp/input/2020/01,/tmp/input/2020/02
/tmp/input/2019/10/29,/tmp/input/2019/10/30,/tmp/input/2019/10/31,/tmp/input/2019/11/01,/tmp/input/2019/11/02

Incorporer dynamiquement des expressions conditionnelles, des chemins, etc. dans des modèles Pig

--Définissez le mot que vous souhaitez remplacer dans le dictionnaire et la valeur que vous souhaitez remplacer, puis générez un Pig intégré dans le modèle.

• À utiliser lorsque vous souhaitez incorporer une expression conditionnelle complexe ou un chemin à changement dynamique.

main.py

def substitute_condition(template, output, target_word, condition):
    txt = ''
    with open(template) as f:
        lines_one_str = f.read()
        txt = lines_one_str.replace(target_word, condition)
    with open(output, mode='w') as f:
        f.write(txt)

def translate(template: str, output: str, d: {str, str}):
    for i, (k, v) in enumerate(d.items()):
        if i == 0:
            substitute_condition(template, output, k, v)
        else:
            substitute_condition(output, output, k, v)
    
d = {'$INPUT': '/tmp/input', '$COND': 'tester|test', '$OUTPUT': '/tmp/output'}
translate('template.pig', 'output.pig', d)

Courir

$ python main.py

template.pig

L = LOAD '$INPUT' USING PigStorage('\t');
F = FILTER L BY note matches '$COND';
FS -rm -r -f -skipTrash $OUTPUT
STORE F INTO '$OUTPUT' USING PigStorage('\t', '-schema');

output.pig

L = LOAD '/tmp/input' USING PigStorage('\t');
F = FILTER L BY note matches 'tester|test';
FS -rm -r -f -skipTrash /tmp/output
STORE F INTO '/tmp/output' USING PigStorage('\t', '-schema');

envoyer un e-mail

def send_mail(subject: str, body: str, from: str, to: str, svr: str, port: str, id: str, password: str):
    msg = MIMEText(body, 'html')
    msg['Subject'] = subject
    msg['From'] = from
    msg['To'] = to

    server = smtplib.SMTP_SSL(svr, port)
    #Pour SSL
    # server = smtplib.SMTP_SSL(svr, port, context=ssl.create_default_context())
    server.login(id, password)
    server.send_message(msg)