根据 Glue 作业状态将文件传输到 S3 存储桶

Question

 I am new to **AWS Glue,** and my aim is to extract  transform and load files uploaded in S3 bucket to RDS instance. Also I need to transfer the files into separate S3 buckets based on the Glue Job status (Success /Failure). There will be more than one file uploaded into the initial S3 bucket. How can I get the name of the files uploaded so that i can transfer those files to appropriate buckets.

第 1 步：将文件上传到 S3 bucket1。第二步：触发lamda函数调用Job1 第 3 步：job1 将文件传输到 S3 bucket2 成功第 4 步：失败时转移到另一个 S3 存储桶

Answer 1

让 lambda 事件触发器监听您正在上传的文件夹文件到 S3 在 lambda 中，使用 AWS Glue API 到运行胶水作业（本质上是 AWS Glue 中的 python 脚本）。

在Glue python脚本中，使用合适的库，比如pymysql等。作为与 python 脚本打包在一起的外部库。

执行从 S3 到 RDS 表的数据加载操作。如果你是使用 Aurora Mysql，那么 AWS 提供了一个不错的功能 "load from S3"，这样就可以直接加载文件到表中（您可能需要在参数组/IAM 角色）。

调用胶水作业的 Lambda 脚本：

s3 = boto3.client('s3')
glue = boto3.client('glue')

def lambda_handler(event, context):
    gluejobname="<YOUR GLUE JOB NAME>"

    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)
    raise e

胶水脚本：

import mysql.connector
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.context import DynamicFrame
from awsglue.transforms import *
from pyspark.sql.types import StringType
from pyspark.sql.types import DateType
from pyspark.sql.functions import unix_timestamp, from_unixtime
from pyspark.sql import SQLContext

# Create a Glue context
glueContext = GlueContext(SparkContext.getOrCreate())

url="<RDS URL>"
uname="<USER NAME>"
pwd="<PASSWORD>"
dbase="DBNAME"


def connect():
    conn = mysql.connector.connect(host=url, user=uname, password=pwd, database=dbase)
    cur = conn.cursor()
    return cur, conn

def create_stg_table():
    cur, conn = connect()
    createStgTable1 = <CREATE STAGING TABLE IF REQUIRED>

    loadQry = "LOAD DATA FROM S3 PREFIX 'S3://PATH FOR YOUR CSV' REPLACE INTO TABLE <DB.TABLENAME> FIELDS TERMINATED BY '|' LINES TERMINATED BY '\n' IGNORE 1 LINES (@var1, @var2, @var3, @var4, @var5, @var6, @var7, @var8) SET ......;"
    cur.execute(createStgTable1)
    cur.execute(loadQry)
    conn.commit()
    conn.close()

然后您可以创建一个 cloudwatch 警报，其中检查粘合作业状态，并根据状态在 S3 之间执行文件复制操作。我们在生产中有类似的设置。

此致

尤瓦

根据 Glue 作业状态将文件传输到 S3 存储桶

Transfering files into S3 buckets based on Glue job status

amazon-s3

amazon-web-services

aws-lambda

aws-glue

调用胶水作业的 Lambda 脚本：

胶水脚本：