AWS Glue：从Timestamp字段创建日期分区

To get the dt=2022-01-01 partition level in your S3 structure, you’ll need to create a new dt column by extracting the date from your timestamp field, then include this column in your partition keys when writing the data. Here’s how to modify your existing Glue script:

Step-by-Step Code Changes #

1. Add the dt Column #

Insert this block right after your ApplyMapping_node2 step. We’ll use Spark DataFrames for straightforward date handling (best for large datasets):

# Convert Glue DynamicFrame to Spark DataFrame
spark_df = ApplyMapping_node2.toDF()

# Import Spark functions for date formatting
from pyspark.sql.functions import date_format, col

# Create 'dt' column with date in 'yyyy-MM-dd' format from timestamp
spark_df_with_dt = spark_df.withColumn("dt", date_format(col("timestamp"), "yyyy-MM-dd"))

# Convert back to Glue DynamicFrame
DynamicFrame_with_dt = glueContext.create_dynamic_frame.from_df(
    spark_df_with_dt,
    glueContext,
    "DynamicFrame_with_dt"
)

2. Update the Write Step to Include dt in Partitions #

Modify your S3bucket_node3 code to add "dt" to the partitionKeys list:

# Script generated for node S3 bucket (updated)
S3bucket_node3 = glueContext.write_dynamic_frame.from_options(
    frame=DynamicFrame_with_dt,  # Use the frame with our new 'dt' column
    connection_type="s3",
    format="glueparquet",
    connection_options={
        "path": "s3://target-data",
        "partitionKeys": ["id", "version", "dt"],  # Add 'dt' here
    },
    format_options={"compression": "gzip"},
    transformation_ctx="S3bucket_node3",
)

Alternative: Glue Native Map Transform (No Spark Conversion) #

If you prefer working directly with Glue DynamicFrames, use the Map transform to add the dt column:

from awsglue.transforms import Map

# Function to extract date from timestamp (Python datetime object)
def add_dt_column(record):
    record["dt"] = record["timestamp"].strftime("%Y-%m-%d")
    return record

# Apply the function to create the 'dt' column
DynamicFrame_with_dt = Map.apply(
    frame=ApplyMapping_node2,
    f=add_dt_column,
    transformation_ctx="DynamicFrame_with_dt"
)

Then update the write step’s partitionKeys as shown above.

Full Modified Script #

Here’s the complete updated code using the Spark-based approach:

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 pyspark.sql.functions import date_format, col  # Added import

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

# Script generated for node S3 bucket
S3bucket_node1 = glueContext.create_dynamic_frame.from_options(
    format_options={},
    connection_type="s3",
    format="parquet",
    connection_options={"paths": ["s3://my-data"], "recurse": True},
    transformation_ctx="S3bucket_node1",
)

# Script generated for node ApplyMapping
ApplyMapping_node2 = ApplyMapping.apply(
    frame=S3bucket_node1,
    mappings=[
        ("timestamp", "timestamp", "timestamp", "timestamp"),
        ("id", "string", "id", "string"),
        ("version", "string", "version", "string"),
    ],
    transformation_ctx="ApplyMapping_node2",
)

# New step: Create 'dt' column from timestamp
spark_df = ApplyMapping_node2.toDF()
spark_df_with_dt = spark_df.withColumn("dt", date_format(col("timestamp"), "yyyy-MM-dd"))
DynamicFrame_with_dt = glueContext.create_dynamic_frame.from_df(
    spark_df_with_dt,
    glueContext,
    "DynamicFrame_with_dt"
)

# Script generated for node S3 bucket (updated partition keys)
S3bucket_node3 = glueContext.write_dynamic_frame.from_options(
    frame=DynamicFrame_with_dt,
    connection_type="s3",
    format="glueparquet",
    connection_options={
        "path": "s3://target-data",
        "partitionKeys": ["id", "version", "dt"],
    },
    format_options={"compression": "gzip"},
    transformation_ctx="S3bucket_node3",
)

job.commit()

Key Notes #

• The date_format function ensures your date uses the yyyy-MM-dd format, which is optimal for S3 partitioning.
• Using Spark DataFrames is recommended for large datasets since it leverages distributed processing.
• After running the modified job, your S3 structure will be id=1/version=2/dt=2022-01-01/ as desired.

内容的提问来源于stack exchange，提问作者Shlomi Schwartz