AWS Glue job insert datetime字段为空（红移）

Question

我是AWS Glue的新手。我正在尝试通过Glue作业插入redshift表，它有S3爬虫来读取csv文件和redshift映射爬虫用于表方案。

下面的作业正在尝试运行，其中create\u日期从S3插入到时间戳的红移列中。值始终为空。

粘合作业：

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: [TempDir, JOB_NAME]
args = getResolvedOptions(sys.argv, ['TempDir','JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
## @type: DataSource
## @args: [database = "salesdatabase", table_name = "sales_timestamp_csv", transformation_ctx = "datasource0"]
## @return: datasource0
## @inputs: []
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "salesdatabase", table_name = "sales_timestamp_csv", transformation_ctx = "datasource0")
## @type: ApplyMapping
## @args: [mapping = [("city", "string", "city", "string"), ("country", "string", "country", "string"), ("amount", "string", "amount", "string"), ("create_date", "string", "create_date", "timestamp")], transformation_ctx = "applymapping1"]
## @return: applymapping1
## @inputs: [frame = datasource0]
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("city", "string", "city", "string"), ("country", "string", "country", "string"), ("amount", "string", "amount", "string"), ("create_date", "string", "create_date", "timestamp")], transformation_ctx = "applymapping1")
## @type: SelectFields
## @args: [paths = ["country", "amount", "city", "create_date"], transformation_ctx = "selectfields2"]
## @return: selectfields2
## @inputs: [frame = applymapping1]
selectfields2 = SelectFields.apply(frame = applymapping1, paths = ["country", "amount", "city", "create_date"], transformation_ctx = "selectfields2")
## @type: ResolveChoice
## @args: [choice = "MATCH_CATALOG", database = "salesdatabase", table_name = "metricsdb_public_sales_csv", transformation_ctx = "resolvechoice3"]
## @return: resolvechoice3
## @inputs: [frame = selectfields2]
resolvechoice3 = ResolveChoice.apply(frame = selectfields2, choice = "MATCH_CATALOG", database = "salesdatabase", table_name = "metricsdb_public_sales_csv", transformation_ctx = "resolvechoice3")
## @type: ResolveChoice
## @args: [choice = "make_cols", transformation_ctx = "resolvechoice4"]
## @return: resolvechoice4
## @inputs: [frame = resolvechoice3]
resolvechoice4 = ResolveChoice.apply(frame = resolvechoice3, choice = "make_cols", transformation_ctx = "resolvechoice4")
## @type: DataSink
## @args: [database = "salesdatabase", table_name = "metricsdb_public_sales_csv", redshift_tmp_dir = TempDir, transformation_ctx = "datasink5"]
## @return: datasink5
## @inputs: [frame = resolvechoice4]
datasink5 = glueContext.write_dynamic_frame.from_catalog(frame = resolvechoice4, database = "salesdatabase", table_name = "metricsdb_public_sales_csv", redshift_tmp_dir = args["TempDir"], transformation_ctx = "datasink5")
job.commit()

爬网程序架构详细信息：

来自爬虫的S3架构

列名数据类型

1. 城市字符串
2. 国家字符串
3. 金额字符串
4. create_date字符串

爬网程序列名称数据类型中的表架构

1. 城市字符串
2. 国家/地区字符串
3. 金额字符串
4. 创建\u日期时间戳

任何指针

Answer 1

I could resolve this.

Just convert Dynamic Frame to Spark Data Frame and apply transformation.

from pyspark.sql.functions import to_timestamp, col
from awsglue.dynamicframe import DynamicFrame

#Convert dynamic frame to data frame to use standard pyspark functions
data_frame = datasource0.toDF()
data_frame.show()
data_frame = data_frame.withColumn("created_date",
          to_timestamp(col("created_date"),"dd/MM/yy HH:mm"))
data_frame.show()


#Convert back to the dynamic frame
dynamic_frame_write = DynamicFrame.fromDF(data_frame, glueContext, 'dynamic_frame_write')

Just add this frame to datasink5
It worked for me