我是AWS的新手,因此不确定如何去做,

我在s3上有CSV文件,并且已经在RDS上设置了Aurora实例。我无法弄清的事情是如何自动执行数据的批量加载,本质上就像使用AWS Glue之类的LOAD DATA FROM s3一样。

我还将s3的Glue原生对象用于RDS,但是从本质上讲,它是一堆通过JDBC连接插入RDS的插入,对于大型数据集而言,这也非常慢。

我可以在RDS上独立运行命令来做到这一点,但是我不想这样做,而是想利用Glue。我还考虑了将MySQL连接器用于Python,但Glue本机仅支持Python 2.7,这是我不想使用的东西。

任何帮助将不胜感激。

最佳答案

如上所述,该方法具有S3事件触发器和Lambda作业,监听S3存储桶/对象位置。将文件上传到s3位置后,lambda作业将运行,并且在lambda中,您可以配置为调用AWS Glue作业。这正是我们所做的,并且已经成功上线了。 Lambda的使用寿命为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


要创建Lambda函数,请转到AWS Lambdra->从头开始创建新函数->为事件选择S3,然后配置S3存储桶位置以及所需的前缀。然后复制粘贴上面的代码示例,内联代码区域,并根据需要配置胶粘作业名称。请确保您具有所有必需的IAM角色/访问设置。

胶粘作业应具有连接到您的Aurora的功能,然后您可以使用Aurora提供的“ LOAD FROM S3 .....”命令。确保根据需要完成所有参数组设置/配置。

让我知道是否有任何问题。

更新:从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()

10-05 22:23
查看更多