Блог

У меня есть следующий сценарий. Я попытался выполнить некоторые другие вопросы и ответы о том, как преобразовать строку в временную метку, но по какой-то причине она не выводится в мою корзину s3. Без шагов, пытающихся преобразовать dob столбец в временную метку, он будет выводиться в мою корзину. Но с помощью шагов задание завершается, но ничего не происходит. Если я использую новую Glue studio и попытаюсь изменить тип столбца со string на timestamp там, все значения столбца dob будут равны нулю.

 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 to_timestamp

## @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)
## @type: DataSource
## @args: [database = "test_db", table_name = "load", transformation_ctx = "DataSource0"]
## @return: DataSource0
## @inputs: []
DataSource0 = glueContext.create_dynamic_frame.from_catalog(database = "test_db", table_name = "load", transformation_ctx = "DataSource0")
# Convert to data frame and perform ETL
dataFrame = DataSource0.toDF().withColumn("dob", to_timestamp("dob","DD/MM/YYY"))
# Convert back to a dynamic frame
editedData = DynamicFrame.fromDF(dataFrame, glueContext, "editedData")
## @type: ApplyMapping
## @args: [mappings = [("id", "long", "id", "long"), ("first_name", "string", "firstName", "string"), ("last_name", "string", "lastName", "string"), ("email", "string", "email", "string"), ("gender", "string", "gender", "string"), ("dob", "timestamp", "dob", "timestamp"), ("phone", "string", "phone", "string"), ("partner_name", "string", "partnerName", "string")], transformation_ctx = "Transform0"]
## @return: Transform0
## @inputs: [frame = DataSource0]
Transform0 = ApplyMapping.apply(frame = editedData, mappings = [("id", "long", "id", "long"), ("first_name", "string", "firstName", "string"), ("last_name", "string", "lastName", "string"), ("email", "string", "email", "string"), ("gender", "string", "gender", "string"), ("dob", "timestamp", "dob", "timestamp"), ("phone", "string", "phone", "string"), ("partner_name", "string", "partnerName", "string")], transformation_ctx = "Transform0")
## @type: DataSink
## @args: [connection_type = "s3", format = "csv", connection_options = {"path": "s3://my-bucket/processed/", "partitionKeys": []}, transformation_ctx = "DataSink0"]
## @return: DataSink0
## @inputs: [frame = Transform0]
DataSink0 = glueContext.write_dynamic_frame.from_options(frame = Transform0, connection_type = "s3", format = "csv", connection_options = {"path": "s3://my-bucket/processed/", "partitionKeys": ["partnerName"]}, transformation_ctx = "DataSink0")
job.commit()