使用AWS Glue Job在Redshift中导入数据时添加时间戳列 [英] Adding timestamp column in importing data in redshift using AWS Glue Job
问题描述
我想知道在由AWS Glue Job加载时是否可以在表中添加时间戳列.
I would like to know if it is possible to add a timestamp column in a table when it is loaded by an AWS Glue Job.
第一种情况:
A列|栏B |时间戳
Column A | Column B| TimeStamp
A | 2 | 2018-06-03 23:59:00.0
A|2|2018-06-03 23:59:00.0
当抓取工具"更新数据目录中的表并再次运行作业时,该表将在表中添加带有新时间戳记的新数据.
When a Crawler updates the table in the data catalog and run the job again, the table will add the new data in the table with a new time stamp..
A列|栏B |时间戳
Column A | Column B| TimeStamp
A | 4 | 2018-06-04 05:01:31.0
A|4|2018-06-04 05:01:31.0
B | 8 | 2018-06-04 06:02:31.0
B|8|2018-06-04 06:02:31.0
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)
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "sampledb", table_name = "abs", transformation_ctx = "datasource0")
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("ColumnA", "char", "ColumnA", "char"), ("ColumnB", "char", "ColumnB", "char")], transformation_ctx = "applymapping1")
resolvechoice2 = ResolveChoice.apply(frame = applymapping1, choice = "make_cols", transformation_ctx = "resolvechoice2")
dropnullfields3 = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = "dropnullfields3")
datasink4 = glueContext.write_dynamic_frame.from_jdbc_conf(frame = dropnullfields3, catalog_connection = "TESTDB", connection_options = {"dbtable": "TABLEA", "database": "anasightprd01"}, redshift_tmp_dir = args["TempDir"], transformation_ctx = "datasink4")
推荐答案
将DynamicFrame转换为spark的DataFrame,添加具有当前时间戳的新列,然后在写入之前将其转换回DynamicFrame.
Convert DynamicFrame to spark's DataFrame, add a new column with current timestamp and then convert it back to DynamicFrame before writing.
import org.apache.spark.sql.functions._
...
val timestampedDf = dropnullfields3.toDF().withColumn("TimeStamp", current_timestamp())
val timestamped4 = DynamicFrame(timestampedDf, glueContext)
这是您的Python代码的外观:
Here how your Python code should look like:
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext, DynamicFrame
from awsglue.job import Job
from pyspark.sql.functions import current_timestamp
## @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)
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "sampledb", table_name = "abs", transformation_ctx = "datasource0")
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("ColumnA", "char", "ColumnA", "char"), ("ColumnB", "char", "ColumnB", "char")], transformation_ctx = "applymapping1")
resolvechoice2 = ResolveChoice.apply(frame = applymapping1, choice = "make_cols", transformation_ctx = "resolvechoice2")
dropnullfields3 = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = "dropnullfields3")
# add TimeStamp column
timestampedDf = dropnullfields3.toDF().withColumn("TimeStamp", current_timestamp())
timestamped4 = DynamicFrame.fromDF(timestampedDf, glueContext, "timestampedDf")
datasink4 = glueContext.write_dynamic_frame.from_jdbc_conf(frame = timestamped4, catalog_connection = "TESTDB", connection_options = {"dbtable": "TABLEA", "database": "anasightprd01"}, redshift_tmp_dir = args["TempDir"], transformation_ctx = "datasink4")
这篇关于使用AWS Glue Job在Redshift中导入数据时添加时间戳列的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!