Автоматизировать массовую загрузку данных из s3 в экземпляр Aurora MySQL RDS - PullRequest
Новая
       8

Автоматизировать массовую загрузку данных из s3 в экземпляр Aurora MySQL RDS

0 голосов
/ 11 апреля 2019

Я относительно новичок в AWS, поэтому не знаю, как это сделать,

У меня есть файлы CSV на s3, и я уже настроил экземпляр Aurora на RDS.То, что я не могу понять, это как автоматизировать массовую загрузку данных, по сути, что-то вроде LOAD DATA FROM s3 с использованием чего-то вроде AWS Glue.

Я также использовал нативную вещь Glue изот s3 до RDS, но это, по сути, куча вставок в RDS по соединению JDBC, которое также очень медленное для больших наборов данных.

Я могу сделать это независимо, выполнив команду на RDS, но я не хочусделайте это и хотите использовать клей.Я также рассмотрел использование коннектора MySQL для Python, но Glue изначально поддерживает только Python 2.7, который я не хочу использовать.

Любая помощь будет принята с благодарностью.

1 Ответ

1 голос
/ 15 апреля 2019

Подход, как указано выше, имеет триггер события S3 и лямбда-задание, прослушивающие область памяти / объект s3.Как только файл будет загружен в папку s3, лямбда-задание будет запущено, и в лямбда-режиме вы сможете настроить вызов AWS Glue.Это именно то, что мы сделали и успешно ушли вживую.Лямбда имеет 15-минутный срок службы, и для запуска / запуска клея может потребоваться меньше минуты.

Для примера ознакомьтесь с примером источника. 

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

Для создания лямбда-функции перейдите в AWS Lambdra-> Создать новую функцию из Scratch-> Выберите S3 для события изатем настройте местоположения S3 bucket, при необходимости префиксы.Затем скопируйте, вставьте приведенный выше пример кода, встроенную область кода и настройте имя задания клея по мере необходимости.Убедитесь, что у вас есть все необходимые роли IAM / настройка доступа.

В задании на склейку должно быть предусмотрено подключение к вашей Aurora, и тогда вы можете использовать команду «LOAD FROM S3 .....», предоставленную Aurora.Убедитесь, что все настройки / конфигурации группы параметров выполнены по мере необходимости.

Дайте мне знать, если возникнут какие-либо проблемы.

ОБНОВЛЕНИЕ: Фрагмент кода SAMPLE для 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, col4=@var4;"
cur.execute(createStgTable1)
cur.execute(createStgTable2)
cur.execute(loadQry)
conn.commit()
conn.close()
...