ETL Pipeline: Načítání dat z AWS S3 pomocí AWS Glue

Obsah

Řešení a zjednodušený diagram architektury
AWS CDK skript (TypeScript)
AWS Glue ETL úloha (PySpark)

Tento pipeline demonstruje, jak vytvořit ETL proces pomocí AWS Glue pro extrakci dat z Amazon S3, jejich transformaci a načtení do cílového úložiště dat (což může být jiný S3 bucket, datový sklad jako Amazon Redshift nebo databáze). Systém využívá bezserverovou architekturu AWS Glue k poskytnutí škálovatelné a spravované ETL služby.

Použitím AWS Glue ve spojení s Amazon S3 mohou vývojáři vytvořit robustní, bezserverové řešení pro zpracování a analýzu dat. Tento přístup ukazuje sílu a flexibilitu AWS služeb při budování škálovatelných, automatizovaných datových pipeline.

Popis případu použití

V tomto scénáři jsou surová data uložena v S3 bucketu. AWS Glue je nakonfigurován k procházení těchto dat, odvození jejich schématu a vytvoření katalogu dostupných datasetů. Poté je vytvořena AWS Glue ETL úloha pro zpracování těchto dat.

ETL úloha, napsaná v PySpark, čte data z S3, provádí nezbytné transformace (jako je čištění dat, konverze formátu nebo agregace) a pak zapisuje zpracovaná data do cílového umístění. To může být jiný S3 bucket, datový sklad nebo databáze, v závislosti na konkrétních požadavcích.

Tento automatizovaný ETL pipeline je navržen tak, aby byl škálovatelný, schopný zpracovat velké objemy dat využitím distribuovaných výpočetních schopností Apache Spark, který AWS Glue používá na pozadí. Eliminuje potřebu správy infrastruktury, protože AWS Glue je plně spravovaná služba.

Průmyslová odvětví

Toto řešení je použitelné v různých průmyslových odvětvích, která se zabývají zpracováním velkých objemů dat, včetně financí, zdravotnictví, maloobchodu a výroby. Jakákoliv aplikace, která zahrnuje extrakci dat z různých zdrojů, jejich transformaci pro analýzu a načtení do datového úložiště pro dotazování, může těžit z tohoto pipeline. Příklady zahrnují systémy business intelligence, platformy pro analýzu zákazníků a systémy pro zpracování IoT dat.

Zjednodušený diagram architektury

AWS Glue S3 ETL Diagram

Architektura se skládá z:

Amazon S3 Bucket: Ukládá vstupní surová data a zpracovaná výstupní data.
AWS Glue Data Catalog: Obsahuje metadata o datasetech.
AWS Glue Crawler: Odvozuje schéma dat v S3 a naplňuje Data Catalog.
AWS Glue ETL úloha: Zpracovává data pomocí PySpark.
IAM Role: Uděluje nezbytná oprávnění AWS Glue pro přístup k S3 a dalším AWS službám.

AWS CDK skript (TypeScript)

import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as glue from 'aws-cdk-lib/aws-glue';
import * as iam from 'aws-cdk-lib/aws-iam';

export class AwsGlueS3EtlStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // Vytvoření S3 bucketů pro vstupní a výstupní data
    const inputBucket = new s3.Bucket(this, 'InputBucket', {
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const outputBucket = new s3.Bucket(this, 'OutputBucket', {
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    // Vytvoření IAM role pro Glue
    const glueRole = new iam.Role(this, 'GlueETLRole', {
      assumedBy: new iam.ServicePrincipal('glue.amazonaws.com'),
    });

    glueRole.addManagedPolicy(iam.ManagedPolicy.fromAwsManagedPolicyName('service-role/AWSGlueServiceRole'));
    inputBucket.grantRead(glueRole);
    outputBucket.grantReadWrite(glueRole);

    // Vytvoření Glue databáze
    const glueDatabase = new glue.CfnDatabase(this, 'GlueDatabase', {
      catalogId: this.account,
      databaseInput: {
        name: 'mojedatabaze',
      },
    });

    // Vytvoření Glue Crawleru
    const glueCrawler = new glue.CfnCrawler(this, 'GlueCrawler', {
      role: glueRole.roleArn,
      targets: {
        s3Targets: [{ path: inputBucket.bucketName }],
      },
      databaseName: glueDatabase.ref,
      schedule: {
        scheduleExpression: 'cron(0 1 * * ? *)',
      },
    });

    // Vytvoření Glue úlohy
    const glueJob = new glue.CfnJob(this, 'GlueETLJob', {
      command: {
        name: 'glueetl',
        pythonVersion: '3',
        scriptLocation: 's3://' + inputBucket.bucketName + '/etl_script.py',
      },
      role: glueRole.roleArn,
      defaultArguments: {
        '--job-language': 'python',
        '--job-bookmark-option': 'job-bookmark-enable',
      },
      executionProperty: {
        maxConcurrentRuns: 2,
      },
      maxRetries: 0,
      name: 'moje-etl-uloha',
      glueVersion: '3.0',
    });
  }
}

AWS Glue ETL úloha (PySpark)

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Čtení dat z S3
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "mojedatabaze", table_name = "vstupni_data", transformation_ctx = "datasource0")

# Provedení transformací
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("sloupec1", "string", "sloupec1", "string"), ("sloupec2", "long", "sloupec2", "long")], transformation_ctx = "applymapping1")

# Zápis transformovaných dat zpět do S3
datasink2 = glueContext.write_dynamic_frame.from_options(frame = applymapping1, connection_type = "s3", connection_options = {"path": "s3://output-bucket/zpracovana_data"}, format = "parquet", transformation_ctx = "datasink2")

job.commit()