दृष्टिकोण जैसा ऊपर बताया गया है, एक एस 3 इवेंट ट्रिगर और एस 3 बाल्टी/ऑब्जेक्ट लोकेशन पर लैम्ब्डा जॉब सुनना है। जैसे ही फ़ाइल को s3 स्थान पर अपलोड किया जाता है, लैम्ब्डा जॉब चलेगा, और लैम्ब्डा में, आप AWS ग्लू जॉब को कॉल करने के लिए कॉन्फ़िगर कर सकते हैं। ठीक यही हमने किया है और सफलतापूर्वक लाइव हो गया है। लैम्ब्डा में 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 लैम्बड्रा पर जाएं-> स्क्रैच से एक नया फ़ंक्शन बनाएं-> ईवेंट के लिए 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, [email protected];"
cur.execute(createStgTable1)
cur.execute(createStgTable2)
cur.execute(loadQry)
conn.commit()
conn.close()
