Liste des Python-Codes, der bei der Big-Data-Analyse verwendet wird

Umgebung

--Erstellen Sie nach Bedarf eine virtuelle Umgebung.

Erstellen Sie eine virtuelle Umgebung

$ python3 -m venv test

Aktivierung der virtuellen Umgebung

$ source test/bin/activate
(test)$

Deaktivieren Sie die virtuelle Umgebung

(test)$ deactivate
$

String-Verarbeitung

Schneiden Sie eine Zeichenkette aus

• [Startindex: Endzeichennummer (beachten Sie, dass es sich nicht um einen Index handelt)].

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

Flucht

Flucht aus gewellten Klammern

• Entwerfen Sie die wellenförmigen Klammern mit wellenförmigen Klammern.

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

Verzeichnismanipulation

Verzeichnis erstellen

import os

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

class

• Verwenden Sie diese Option, wenn Sie Eigenschaften usw. trennen möchten.

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

main.py

from prop.user_property import UserProperty

user_property = UserProperty({'first_name': 'Ichiro', 'family_name': 'Prüfung'})
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']

Ausführungsergebnis

$ python main.py
Teste Ichiro

subprocess

• Sie können Shell-Befehle von Python mit Unterprozess ausführen.
• Big Data kann nicht analysiert werden, ohne Shell-Befehle auszuführen.

Shell-Befehl ausführen

import subprocess

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

Parallele Ausführung von Shell-Befehlen mit xargs

• Sie müssen shell = True setzen, um Pipes zu verwenden.
• Sie können den Prozess mit subprocess.Popen () empfangen und mit wait () auf den Abschluss der Verarbeitung warten.
• Die nachfolgende Verarbeitung wird erst ausgeführt, wenn der Vorgang abgeschlossen ist.
• Das Beispiel katzt jede Datei in tmp von Python und führt test.sh mit 10 Parallelen aus.
• Verwenden Sie diese Option, wenn Sie die Verarbeitungszeit verkürzen möchten, indem Sie nur die Zwischenverarbeitung parallel ausführen.

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

#Nachbearbeitung

Handhabung der Standardausgabe

• Übergeben Sie stdout und stderr nicht an Pipe, wenn Sie sie nicht benötigen.
• Sie können die Standardausgabe mit python test.py &> log / test.log empfangen.

click

• Sie können einfach Befehle implementieren, die im Terminal ausgeführt werden können, indem Sie auf klicken.
• Implementieren Sie den Befehl mit @ click.command (). --Mehrere Befehle können mit @ click.group () und add_command () implementiert werden.
• Sie können Befehlsargumente mit @ click.option () hinzufügen.

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.py Nachricht.')

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='Geben Sie die Nachricht ein, die Sie anzeigen möchten.')
def hello(**kwargs):
    print(f'Nachricht eingegeben:{kwargs["msg"]}')
    print('click/cmd/hello/cli.py Nachricht.')

$ python cli.py hello -m 'Prüfung'
click/cli.py Nachricht.
Nachricht eingegeben: Test
click/cmd/hello/cli.py Nachricht.

pandas

• Wird bei der Datenverarbeitung verwendet.

tsv gelesen

• Sie können ein Trennzeichen mit Trennzeichen angeben.
• Sie können Spaltennamen mit Namen festlegen.
• Sie können den Datentyp mit dtype angeben. --Set low_memory = False beim Umgang mit großen Dateien.

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)

tsv Ausgabe

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

Ausgabe durch Angabe einer bestimmten Spalte

• Wird verwendet, wenn Sie Spalten anordnen möchten, die für die Analyse erforderlich sind, jedoch nicht benötigt werden.

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

Ausgabe durch Eingrenzen der Anzahl der Fälle

• Zur Probenahme verwendet.

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

Doppelte Zeilenentfernung

df.drop_duplicates()

Verwenden Sie bei der Abfrage doppelte Anführungszeichen

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

Fehlerbehandlung

Python-Skript töten

• Verwenden Sie diese Option, wenn Sie ein Python-Skript beenden möchten, wenn ein Fehler auftritt.

import sys

sys.exit(1)

Überprüfung der Dateiexistenz

―― Wird verwendet, um zu überprüfen, ob Sie über die erforderlichen Eingaben verfügen, bevor Sie eine Datenanalyse durchführen.

import os

if os.path.exists():
    print('Die Datei existiert. Führt die nachfolgende Verarbeitung durch.')
else:
    print('Die Datei existiert nicht. Der Prozess endet.')
    sys.exit(1)

Protokollierung

--Verwenden Sie die Standardmodulprotokollierung von Python.

• Ein Protokollierungsbeispiel mit der folgenden Konfiguration wird beschrieben.

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

main.py

#Selbstgemachtes Modul
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("Starten Sie die Skriptausführung")

if __name__ == "__main__":
    init()
    #Rufen Sie die Funktion Ihres eigenen Moduls auf
    sub.hello()

module/sub.py

from logging import getLogger

logger = getLogger(__name__)

def hello():
    print('hello! this is sub module.')
    logger.debug('Ausgabe vom Submodul')

$ python main.py
[20-06-25 14:20:56] __main__ DEBUG:Starten Sie die Skriptausführung
hello! this is sub module.
[20-06-25 14:20:56] module.sub DEBUG:Ausgabe vom Submodul

$ head log/test.log
PID:15171 [20-06-25 14:20:56] __main__ DEBUG:Starten Sie die Skriptausführung
PID:15171 [20-06-25 14:20:56] module.sub DEBUG:Ausgabe vom Submodul

Andere

Holen Sie sich die Anzahl der Dateien

--Kann in einer Zeile abgerufen werden, ohne einen Shell-Befehl auszuführen.

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

Machen Sie die Datei zu einer einzeiligen Zeichenfolge

• Wird verwendet, wenn mehrere Schlüsselwörter unter der ODER-Bedingung in einer Zeile zusammengefasst werden.

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
Prüfung
test
Text
text
Geschmack
$ python main.py
Prüfung|test|Text|text|Geschmack|taste

Generieren Sie dynamisch ein Verzeichnis für die Datumspartition

• Bei der Datenanalyse gibt es viele Situationen, in denen Daten im Wert von n Monaten und Daten im Wert von n Tagen gelesen werden. Daher werden sie in solchen Fällen verwendet.

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-Bereiten Sie ein Verzeichnis für 11 bis 4 Monate vor
print(out_term(2019, 11, 4, '/tmp/input/'))
# 2019-11-Bereiten Sie ein Verzeichnis vor, das 5 Tage ab 02 zurückreicht
print(out_reverse_term_by_day(datetime.date(2019, 11, 2), 5, '/tmp/input/'))

Ausführungsergebnis

$ 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

Betten Sie bedingte Ausdrücke, Pfade usw. dynamisch in Pig-Vorlagen ein

• Definieren Sie das Wort, das Sie im Wörterbuch ersetzen möchten, und den Wert, den Sie ersetzen möchten, und generieren Sie ein in die Vorlage eingebettetes Schwein.
• Verwenden Sie diese Option, wenn Sie einen komplizierten bedingten Ausdruck oder einen sich dynamisch ändernden Pfad einbetten möchten.

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': 'Prüfung|test', '$OUTPUT': '/tmp/output'}
translate('template.pig', 'output.pig', d)

Lauf

$ 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 'Prüfung|test';
FS -rm -r -f -skipTrash /tmp/output
STORE F INTO '/tmp/output' USING PigStorage('\t', '-schema');

E-Mail senden

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)
    #Für SSL
    # server = smtplib.SMTP_SSL(svr, port, context=ssl.create_default_context())
    server.login(id, password)
    server.send_message(msg)