Mengotomatiskan pemuatan massal data dari s3 ke instans Aurora MySQL RDS

Pendekatannya seperti yang dinyatakan di atas, memiliki pemicu peristiwa S3 dan pekerjaan lambda yang mendengarkan di lokasi bucket/objek s3. Segera setelah file diunggah ke lokasi s3, tugas lambda akan berjalan, dan di lambda, Anda dapat mengonfigurasi untuk memanggil tugas Lem AWS. Inilah tepatnya yang telah kami lakukan dan telah berhasil ditayangkan. Lambda memiliki masa pakai 15 menit, dan dibutuhkan kurang dari satu menit untuk memicu/memulai pekerjaan Lem.

Silakan temukan di sini sumber sampel untuk referensi.

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

Untuk membuat fungsi Lambda, buka AWS Lambdra->Create a new function from Scratch->Select S3 for event, lalu konfigurasikan lokasi bucket S3, awalan sesuai kebutuhan. Kemudian salin tempel contoh kode di atas, area kode sebaris, dan konfigurasikan nama pekerjaan lem sesuai kebutuhan. Pastikan Anda memiliki semua peran IAM/penyiapan akses yang diperlukan.

Pekerjaan lem harus memiliki ketentuan untuk terhubung ke Aurora Anda, dan kemudian Anda dapat menggunakan perintah "LOAD FROM S3....." yang disediakan oleh Aurora. Pastikan semua pengaturan/konfigurasi grup parameter dilakukan sesuai kebutuhan.

Beri tahu saya jika ada masalah.

PERBARUI:SAMPEL cuplikan kode untuk 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()