LOADING CSV to AWS DYNAMO DB USING AWS GLUE скачать в хорошем качестве

LOADING CSV to AWS DYNAMO DB USING AWS GLUE

LOADING CSV to AWS DYNAMO DB USING AWS GLUE #awsglue #AWSGlue #DynamoDBLoad #CSVtoDynamoDB #S3Pipeline #ETLFlow #ServerlessData #PartitionKeyRitual #MovementMetadata #CloudSanctuary #PySparkBlessings #IAMAlignment #SchemaMapping #DeploymentHeartbeat #OnboardingScrolls #TechnicalRituals #DataEmpowerment #ResilientETL #LegacyInCode 🛠️ Step-by-Step: Load CSV from S3 to DynamoDB Using AWS Glue 1. Prepare Your CSV in S3 • Upload your CSV file to an S3 bucket. • Ensure it has headers that match the attribute names you want in DynamoDB. • Include a partition key column—this is mandatory for DynamoDB. 2. Create a Glue Crawler (Optional but Recommended) • Use a Glue Crawler to scan your CSV and create a Data Catalog table. • This helps Glue infer the schema and simplifies your script. 3. Create a Script-Based Glue Job • In AWS Glue, create a new job and choose Spark (Python). • Use PySpark to read the CSV and write to DynamoDB. Here's the code 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 from awsglue.dynamicframe import DynamicFrame from pyspark.sql.functions import col, monotonically_increasing_id Initialize job args = getResolvedOptions(sys.argv, ['JOB_NAME']) sc = SparkContext() glueContext = GlueContext(sc) spark = glueContext.spark_session job = Job(glueContext) job.init(args['JOB_NAME'], args) Load data from S3 AmazonS3_node = glueContext.create_dynamic_frame.from_options( format_options={ "quoteChar": "\"", "withHeader": True, "separator": ",", "optimizePerformance": False }, connection_type="s3", format="csv", connection_options={ "paths": ["s3://aware-it-test01"], "recurse": True }, transformation_ctx="AmazonS3_node" ) Convert to DataFrame for transformation df = AmazonS3_node.toDF() Ensure 'id' exists and is cast to Number (LongType) if 'id' not in df.columns: df = df.withColumn("id", monotonically_increasing_id()) df = df.withColumn("id", col("id").cast("long")) Convert back to DynamicFrame dyf = DynamicFrame.fromDF(df, glueContext, "dyf") Write to DynamoDB glueContext.write_dynamic_frame.from_options( frame=dyf, connection_type="dynamodb", connection_options={ "dynamodb.output.tableName": "Products3", "dynamodb.throughput.write.percent": "1.0" } ) job.commit()