L'approccio è come indicato sopra, dispone di un trigger di eventi S3 e di un processo lambda in ascolto sulla posizione dell'oggetto/del bucket s3. Non appena un file viene caricato nella posizione s3, il lavoro lambda verrà eseguito e in lambda puoi configurare per chiamare un lavoro AWS Glue. Questo è esattamente ciò che abbiamo fatto ed è andato online con successo. Lambda ha una durata di 15 minuti e dovrebbe essere necessario meno di un minuto per attivare/avviare un lavoro di incollaggio.
Si prega di trovare qui una fonte di esempio per riferimento.
from __future__ import print_function
import json
import boto3
import time
import urllib
print('Loading function')
s3 = boto3.client('s3')
glue = boto3.client('glue')
def lambda_handler(event, context):
gluejobname="your-glue-job-name here"
try:
runId = glue.start_job_run(JobName=gluejobname)
status = glue.get_job_run(JobName=gluejobname, RunId=runId['JobRunId'])
print("Job Status : ", status['JobRun']['JobRunState'])
except Exception as e:
print(e)
print('Error getting object {} from bucket {}. Make sure they exist '
'and your bucket is in the same region as this '
'function.'.format(source_bucket, source_bucket))
raise e
Per creare una funzione Lambda, vai su AWS Lambdra->Crea una nuova funzione da Scratch->Seleziona S3 per l'evento, quindi configura le posizioni dei bucket S3, i prefissi come richiesto. Quindi copia incolla l'esempio di codice sopra, l'area del codice inline e configura il nome del lavoro di colla come necessario. Assicurati di avere tutti i ruoli IAM/l'impostazione dell'accesso richiesti.
Il lavoro di incollaggio dovrebbe prevedere la connessione all'Aurora, quindi puoi utilizzare il comando "LOAD FROM S3....." fornito da Aurora. Assicurati che tutte le impostazioni/configurazioni del gruppo di parametri siano eseguite come necessario.
Fammi sapere se ci sono problemi.
UPDATE:Snippet di codice SAMPLE per LOAD FROM S3:
conn = mysql.connector.connect(host=url, user=uname, password=pwd, database=dbase)
cur = conn.cursor()
cur, conn = connect()
createStgTable1 = "DROP TABLE IF EXISTS mydb.STG_TABLE;"
createStgTable2 = "CREATE TABLE mydb.STG_TABLE(COL1 VARCHAR(50) NOT NULL, COL2 VARCHAR(50), COL3 VARCHAR(50), COL4 CHAR(1) NOT NULL);"
loadQry = "LOAD DATA FROM S3 PREFIX 's3://<bucketname>/folder' REPLACE INTO TABLE mydb.STG_TABLE FIELDS TERMINATED BY '|' LINES TERMINATED BY '\n' IGNORE 1 LINES (@var1, @var2, @var3, @var4) SET col1= @var1, col2= @var2, col3= @var3, [email protected];"
cur.execute(createStgTable1)
cur.execute(createStgTable2)
cur.execute(loadQry)
conn.commit()
conn.close()
